mirror of
https://github.com/Strategic-Automation/violin.git
synced 2026-08-14 12:33:37 +02:00
feat(guard): replace custom shell regexes with bashlex AST parsing
This commit is contained in:
@@ -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()))
|
||||
@@ -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:
|
||||
|
||||
@@ -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"(?<![\w.])(?:\d{1,3}\.){3}\d{1,3}(?![\w.])")
|
||||
_DOMAIN_RE = re.compile(
|
||||
r"(?<![\w.-])(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,63}(?![\w.-])",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_URL_RE = re.compile(r"\b(?:https?|ftp|wss?|file)://[^\s'\"<>]+", 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
|
||||
|
||||
@@ -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"(?<![\w.])(?:\d{1,3}\.){3}\d{1,3}(?![\w.])")
|
||||
_DOMAIN_RE = re.compile(
|
||||
r"(?<![\w.-])(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,63}(?![\w.-])",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_URL_RE = re.compile(r"\b(?:https?|ftp|wss?|file)://[^\s'\"<>]+", 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",
|
||||
}
|
||||
)
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
@@ -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" },
|
||||
|
||||
Reference in New Issue
Block a user