mirror of
https://github.com/Strategic-Automation/violin.git
synced 2026-08-14 12:33:37 +02:00
@@ -4,6 +4,9 @@ on:
|
||||
push:
|
||||
pull_request:
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
|
||||
jobs:
|
||||
test:
|
||||
strategy:
|
||||
|
||||
@@ -4,6 +4,10 @@ on:
|
||||
branches: [master, dev]
|
||||
pull_request:
|
||||
branches: [master, dev]
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
|
||||
jobs:
|
||||
guard-check:
|
||||
runs-on: ubuntu-latest
|
||||
|
||||
@@ -4,6 +4,10 @@ on:
|
||||
branches: [master]
|
||||
pull_request:
|
||||
branches: [master]
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
|
||||
jobs:
|
||||
yaml-lint:
|
||||
runs-on: ubuntu-latest
|
||||
@@ -30,4 +34,4 @@ jobs:
|
||||
sys.exit(1)
|
||||
else:
|
||||
print(f'All {len(files)} YAML files valid')
|
||||
"
|
||||
"
|
||||
|
||||
@@ -1,5 +1,29 @@
|
||||
# Changelog
|
||||
|
||||
## 2.0.6
|
||||
|
||||
- Resolved the current CodeQL standard quality findings by making intentional exception fallbacks explicit and removing unused test and hypothesis variables.
|
||||
|
||||
## 2.0.5
|
||||
|
||||
- Restored exact-repeat detection for execution history entries with receipt paths and added unambiguous command-length metadata while retaining compatibility with existing history files.
|
||||
|
||||
## 2.0.4
|
||||
|
||||
- Hard-blocked callback and research endpoints when supplied as primary assessment targets while preserving their approved secondary-only use, including burst execution.
|
||||
|
||||
## 2.0.3
|
||||
|
||||
- Fixed raw-terminal compound-command classification so every pipeline, logical, semicolon, and newline segment is checked independently, and package/source exemptions require every URL in the segment to use an approved source host.
|
||||
|
||||
## 2.0.2
|
||||
|
||||
- Restricted all GitHub Actions workflow tokens to read-only repository contents, resolving the three least-privilege code-scanning alerts without changing workflow behavior.
|
||||
|
||||
## 2.0.1
|
||||
|
||||
- Upgraded the pytest development dependency to 9.0.3 or later to address CVE-2025-71176 insecure temporary-directory handling.
|
||||
|
||||
## 2.0.0
|
||||
|
||||
- Added model-visible `violin_status` diagnostics, phase-aware 10/20-command sync windows, a 350-iteration profile budget, and atomic `violin_review_batch` reconciliation with optional receipt-backed finding output.
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
# violin - supervised agentic Hermes pentest profile
|
||||
name: violin
|
||||
version: 2.0.0
|
||||
version: 2.0.6
|
||||
description: "A supervised agentic Hermes penetration testing profile for authorised Kali/Parrot-based security assessment, reconnaissance, exploit validation, and reporting workflows."
|
||||
hermes_requires: ">=0.18.0"
|
||||
author: "Violin contributors"
|
||||
|
||||
@@ -169,12 +169,10 @@ def _on_session_finalize_hook(session_id=None, eng_dir=None, **kwargs) -> None:
|
||||
we leave a continuity marker so a fresh session can re-read pending state.
|
||||
"""
|
||||
if eng_dir:
|
||||
try:
|
||||
with contextlib.suppress(Exception):
|
||||
pending = state.has_pending_sync(str(eng_dir))
|
||||
if pending:
|
||||
state.set_heartbeat_pending(
|
||||
str(eng_dir),
|
||||
"session finalized with a pending sync lock; run violin_review_batch",
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
@@ -338,7 +338,6 @@ def check_hypothesis_freshness(
|
||||
"or 'source unavailable' are valid outcomes when truthful."
|
||||
)
|
||||
return result
|
||||
relevant = researched
|
||||
|
||||
# Check for stale hypotheses (no update in 48h)
|
||||
stale = 0
|
||||
|
||||
@@ -126,12 +126,10 @@ def _terminate_pid(pid: int) -> None:
|
||||
check=False,
|
||||
)
|
||||
else:
|
||||
try:
|
||||
with contextlib.suppress(ProcessLookupError):
|
||||
os.killpg(pid, signal.SIGTERM)
|
||||
time.sleep(0.2)
|
||||
os.killpg(pid, signal.SIGKILL)
|
||||
except ProcessLookupError:
|
||||
pass
|
||||
|
||||
|
||||
def _terminate_process(proc: subprocess.Popen) -> None:
|
||||
@@ -144,8 +142,7 @@ def _terminate_process(proc: subprocess.Popen) -> None:
|
||||
proc.wait(timeout=1)
|
||||
return
|
||||
except (OSError, subprocess.TimeoutExpired):
|
||||
pass
|
||||
_terminate_pid(proc.pid)
|
||||
_terminate_pid(proc.pid)
|
||||
if proc.poll() is None:
|
||||
with contextlib.suppress(OSError):
|
||||
proc.kill()
|
||||
@@ -219,13 +216,11 @@ def _monitor_background(
|
||||
status_name = "timed_out"
|
||||
_terminate_process(proc)
|
||||
break
|
||||
try:
|
||||
with contextlib.suppress(OSError):
|
||||
if stdout_path.stat().st_size + stderr_path.stat().st_size > MAX_OUTPUT_BYTES:
|
||||
status_name = "output_limited"
|
||||
_terminate_process(proc)
|
||||
break
|
||||
except OSError:
|
||||
pass
|
||||
time.sleep(0.1)
|
||||
try:
|
||||
exit_code = proc.wait(timeout=5)
|
||||
|
||||
@@ -12,6 +12,10 @@ from pathlib import Path
|
||||
|
||||
from .state import lock_file, resolve_eng_dir
|
||||
|
||||
_COMMAND_MARKER = " | command="
|
||||
_COMMAND_LENGTH_MARKER = " | command_length="
|
||||
_RECEIPT_MARKER = " | receipt="
|
||||
|
||||
|
||||
def _history_path(eng_dir: str | Path) -> Path:
|
||||
return resolve_eng_dir(eng_dir) / "state" / "history.md"
|
||||
@@ -28,9 +32,12 @@ def append_history(
|
||||
path = _history_path(eng_dir)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
stamp = datetime.now(UTC).isoformat().replace("+00:00", "Z")
|
||||
line = f"- {stamp} | phase={phase} | exit_code={exit_code} | command={command}"
|
||||
line = (
|
||||
f"- {stamp} | phase={phase} | exit_code={exit_code} | command={command}"
|
||||
f"{_COMMAND_LENGTH_MARKER}{len(command)}"
|
||||
)
|
||||
if receipt_path:
|
||||
line += f" | receipt={receipt_path}"
|
||||
line += f"{_RECEIPT_MARKER}{receipt_path}"
|
||||
with lock_file(path), path.open("a", encoding="utf-8") as handle:
|
||||
handle.write(line + "\n")
|
||||
|
||||
@@ -43,13 +50,32 @@ def history_contains(eng_dir: str | Path, command: str) -> bool:
|
||||
hist = _history_path(eng_dir)
|
||||
if not hist.exists():
|
||||
return False
|
||||
marker = f" | command={command}"
|
||||
for line in hist.read_text(encoding="utf-8").splitlines():
|
||||
if line.endswith(marker) or f"{marker} | receipt=" in line:
|
||||
if _recorded_command(line) == command:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _recorded_command(line: str) -> str | None:
|
||||
"""Read one command field, including legacy records without a length."""
|
||||
|
||||
if _COMMAND_MARKER not in line:
|
||||
return None
|
||||
payload = line.split(_COMMAND_MARKER, 1)[1]
|
||||
command, marker, metadata = payload.rpartition(_COMMAND_LENGTH_MARKER)
|
||||
if marker:
|
||||
length_text = metadata.split(_RECEIPT_MARKER, 1)[0]
|
||||
try:
|
||||
expected_length = int(length_text)
|
||||
except ValueError:
|
||||
expected_length = -1
|
||||
if expected_length >= 0 and len(command) == expected_length:
|
||||
return command
|
||||
|
||||
command, marker, _receipt = payload.rpartition(_RECEIPT_MARKER)
|
||||
return command if marker else payload
|
||||
|
||||
|
||||
def check_history_staleness(
|
||||
eng_dir: str | Path, command: str, *, allow_pending_repeat: bool = False
|
||||
) -> tuple[list[str], list[str], list[str]]:
|
||||
@@ -77,8 +103,7 @@ def check_history_staleness(
|
||||
# that field exactly instead of using substring matching, which can reject
|
||||
# a command merely because it contains the previous command text.
|
||||
last_line = lines[-1]
|
||||
marker = " | command="
|
||||
recorded_command = last_line.split(marker, 1)[1] if marker in last_line else None
|
||||
recorded_command = _recorded_command(last_line)
|
||||
if recorded_command == command and not allow_pending_repeat:
|
||||
errors.append(
|
||||
f"command appears to be an exact repeat of the last recorded command: {last_line}"
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
name: violin-guard
|
||||
version: "2.0.0"
|
||||
version: "2.0.6"
|
||||
description: Typed scope guards and an execute-and-record boundary with bounded synchronization windows.
|
||||
kind: standalone
|
||||
provides_tools:
|
||||
|
||||
@@ -7,6 +7,7 @@ URL authorities, and ``ipaddress`` for IP/CIDR validation.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import ipaddress
|
||||
import re
|
||||
import shlex
|
||||
@@ -78,9 +79,16 @@ class _TargetPolicy:
|
||||
candidate, self.allowed_networks
|
||||
)
|
||||
|
||||
def is_secondary_only(self, candidate: str) -> bool:
|
||||
return _matches_host(candidate, self.callback_hosts | self.research_hosts)
|
||||
|
||||
def check_primary(self, candidate: str, result: TargetCheckResult) -> None:
|
||||
if self.is_excluded(candidate):
|
||||
result.errors.append(f"excluded target {candidate} must not be touched")
|
||||
elif self.is_secondary_only(candidate):
|
||||
result.errors.append(
|
||||
f"secondary-only endpoint {candidate} must not be used as a primary target"
|
||||
)
|
||||
elif self.is_assessment_target(candidate):
|
||||
return
|
||||
elif _is_ip_network(candidate):
|
||||
@@ -96,9 +104,7 @@ class _TargetPolicy:
|
||||
def check_secondary(self, candidate: str, result: TargetCheckResult) -> None:
|
||||
if self.is_excluded(candidate):
|
||||
result.errors.append(f"excluded target {candidate} must not be touched")
|
||||
elif self.is_assessment_target(candidate) or candidate in (
|
||||
self.callback_hosts | self.research_hosts
|
||||
):
|
||||
elif self.is_assessment_target(candidate) or self.is_secondary_only(candidate):
|
||||
return
|
||||
elif _is_ip_network(candidate):
|
||||
result.errors.append(f"out-of-scope target {candidate} (not present in scope.yaml)")
|
||||
@@ -149,12 +155,10 @@ def normalise_target(value: str) -> str:
|
||||
|
||||
raw = value.strip()
|
||||
raw = re.split(r"\s+\(", raw, maxsplit=1)[0].strip()
|
||||
try:
|
||||
with contextlib.suppress(ValueError):
|
||||
parsed = urlsplit(raw if "://" in raw else f"//{raw}")
|
||||
if parsed.hostname:
|
||||
return parsed.hostname.lower()
|
||||
except ValueError:
|
||||
pass
|
||||
return raw.lower()
|
||||
|
||||
|
||||
@@ -215,12 +219,10 @@ def resolve_target(
|
||||
|
||||
# Extract the requested field from a URL
|
||||
if "://" in target_val and field in ("ip", "host"):
|
||||
try:
|
||||
with contextlib.suppress(ValueError):
|
||||
parsed = urlsplit(target_val)
|
||||
if parsed.hostname:
|
||||
return parsed.hostname
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return target_val
|
||||
|
||||
@@ -286,12 +288,10 @@ def _parse_target_token(token: str) -> str | None:
|
||||
if not raw:
|
||||
return None
|
||||
unbracketed = raw[1:-1] if raw.startswith("[") and raw.endswith("]") else raw
|
||||
try:
|
||||
with contextlib.suppress(ValueError):
|
||||
if "/" in unbracketed:
|
||||
return str(ipaddress.ip_network(unbracketed, strict=False)).lower()
|
||||
return str(ipaddress.ip_address(unbracketed)).lower()
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
try:
|
||||
parsed = urlsplit(raw if raw.startswith("//") or "://" in raw else f"//{raw}")
|
||||
|
||||
@@ -55,11 +55,18 @@ _DOMAIN_RE = re.compile(
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_URL_RE = re.compile(r"\b(?:https?|ftp|wss?|file)://[^\s'\"<>]+", re.IGNORECASE)
|
||||
_KNOWN_SOURCE_HOST_RE = re.compile(
|
||||
r"https?://(?:[^/]*\.)?(?:github\.com|gitlab\.com|bitbucket\.org|"
|
||||
r"pypi\.org|files\.pythonhosted\.org|registry\.npmjs\.org|"
|
||||
r"crates\.io|proxy\.golang\.org|go\.dev)(?::\d+)?(?:/|$)",
|
||||
re.IGNORECASE,
|
||||
_KNOWN_SOURCE_HOSTS = frozenset(
|
||||
{
|
||||
"bitbucket.org",
|
||||
"crates.io",
|
||||
"files.pythonhosted.org",
|
||||
"github.com",
|
||||
"gitlab.com",
|
||||
"go.dev",
|
||||
"proxy.golang.org",
|
||||
"pypi.org",
|
||||
"registry.npmjs.org",
|
||||
}
|
||||
)
|
||||
_NETWORK_PATH_RE = re.compile(r"/(?:dev/)?(?:tcp|udp)/", re.IGNORECASE)
|
||||
_NETWORK_MODULE_RE = re.compile(
|
||||
@@ -123,6 +130,13 @@ def _url_hosts(command: str) -> list[str]:
|
||||
return hosts
|
||||
|
||||
|
||||
def _is_known_source_host(host: str) -> bool:
|
||||
normalized = host.lower().rstrip(".")
|
||||
return any(
|
||||
normalized == known or normalized.endswith(f".{known}") for known in _KNOWN_SOURCE_HOSTS
|
||||
)
|
||||
|
||||
|
||||
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):
|
||||
@@ -163,6 +177,42 @@ def _has_target_literal(command: str) -> bool:
|
||||
return False
|
||||
|
||||
|
||||
def _block_terminal_segment(segment: str) -> str | None:
|
||||
if _NETWORK_PATH_RE.search(segment):
|
||||
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):
|
||||
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)
|
||||
if not is_source_command and url_hosts:
|
||||
return _message("URL detected in a non-package raw terminal command")
|
||||
|
||||
# Public package/source URLs are host-local setup, not assessment traffic.
|
||||
# Keep numeric authorities conservative: a clone/install from an IP may be
|
||||
# an engagement target and must go through the typed guard.
|
||||
if (
|
||||
is_source_command
|
||||
and url_hosts
|
||||
and all(_is_known_source_host(host) for host in url_hosts)
|
||||
and not _IPV4_RE.search(segment)
|
||||
):
|
||||
return None
|
||||
|
||||
if executable not in _LOCAL_COMMANDS and _has_target_literal(segment):
|
||||
return _message("target host literal detected in the raw terminal command")
|
||||
|
||||
if executable in _SCRIPT_INTERPRETERS and _SUSPICIOUS_SCRIPT_RE.search(segment):
|
||||
return _message("assessment script detected in the raw terminal command")
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def block_terminal_command(command: str) -> str | None:
|
||||
"""Return a block message for clearly target-touching raw terminal calls.
|
||||
|
||||
@@ -174,37 +224,9 @@ def block_terminal_command(command: str) -> str | None:
|
||||
if not isinstance(command, str) or not command.strip():
|
||||
return None
|
||||
|
||||
if _NETWORK_PATH_RE.search(command):
|
||||
return _message("network socket path detected in the raw terminal command")
|
||||
|
||||
executable = _first_executable(command)
|
||||
if executable in _SCRIPT_INTERPRETERS and _NETWORK_MODULE_RE.search(command):
|
||||
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.
|
||||
url_hosts = _url_hosts(command)
|
||||
if not _is_package_or_source_command(command) and url_hosts:
|
||||
return _message("URL detected in a non-package raw terminal command")
|
||||
|
||||
# Public package/source URLs are host-local setup, not assessment traffic.
|
||||
# Keep numeric authorities conservative: a clone/install from an IP may be
|
||||
# an engagement target and must go through the typed guard.
|
||||
if (
|
||||
_is_package_or_source_command(command)
|
||||
and url_hosts
|
||||
and _KNOWN_SOURCE_HOST_RE.search(command)
|
||||
and not _IPV4_RE.search(command)
|
||||
):
|
||||
return None
|
||||
|
||||
if executable not in _LOCAL_COMMANDS and _has_target_literal(command):
|
||||
return _message("target host literal detected in the raw terminal command")
|
||||
|
||||
if executable in _SCRIPT_INTERPRETERS and _SUSPICIOUS_SCRIPT_RE.search(command):
|
||||
return _message("assessment script detected in the raw terminal command")
|
||||
|
||||
for segment in _COMMAND_SPLIT_RE.split(command):
|
||||
if message := _block_terminal_segment(segment):
|
||||
return message
|
||||
return None
|
||||
|
||||
|
||||
|
||||
+2
-2
@@ -1,13 +1,13 @@
|
||||
[project]
|
||||
name = "violin"
|
||||
version = "2.0.0"
|
||||
version = "2.0.6"
|
||||
description = "Supervised agentic Hermes penetration-testing profile"
|
||||
requires-python = ">=3.11"
|
||||
dependencies = ["filelock>=3.13,<4"]
|
||||
|
||||
[dependency-groups]
|
||||
dev = [
|
||||
"pytest>=8.0,<9",
|
||||
"pytest>=9.0.3,<10",
|
||||
"pyyaml>=6.0,<7",
|
||||
"ruff>=0.11,<0.12",
|
||||
]
|
||||
|
||||
@@ -2,22 +2,38 @@
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from plugins.violin_guard.history import check_history_staleness
|
||||
from plugins.violin_guard.history import append_history, check_history_staleness, history_contains
|
||||
|
||||
|
||||
def test_history_deduplication_compares_the_recorded_command_field(tmp_path: Path) -> None:
|
||||
history = tmp_path / "state" / "history.md"
|
||||
history.parent.mkdir()
|
||||
history.write_text(
|
||||
"- 2026-07-14T10:00:00Z | phase=RECON | exit_code=0 | command=echo done\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
for suffix in ("", " | receipt=evidence/executions/test.json"):
|
||||
history.write_text(
|
||||
f"- 2026-07-14T10:00:00Z | phase=RECON | exit_code=0 | command=echo done{suffix}\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
errors, _, _ = check_history_staleness(tmp_path, "echo")
|
||||
assert not errors
|
||||
errors, _, _ = check_history_staleness(tmp_path, "echo")
|
||||
assert not errors
|
||||
|
||||
errors, _, _ = check_history_staleness(tmp_path, "echo done")
|
||||
errors, _, _ = check_history_staleness(tmp_path, "echo done")
|
||||
assert errors
|
||||
|
||||
|
||||
def test_written_history_uses_command_length_for_unambiguous_receipt_parsing(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
command = "printf 'value | receipt=fake | command_length=1'"
|
||||
append_history(tmp_path, command, "RECON", 0, "evidence/executions/test.json")
|
||||
|
||||
errors, _, _ = check_history_staleness(tmp_path, command)
|
||||
assert errors
|
||||
assert history_contains(tmp_path, command)
|
||||
|
||||
errors, _, infos = check_history_staleness(tmp_path, command, allow_pending_repeat=True)
|
||||
assert not errors
|
||||
assert any("pending batch" in info for info in infos)
|
||||
|
||||
|
||||
def test_malformed_history_line_does_not_create_a_false_repeat(tmp_path: Path) -> None:
|
||||
|
||||
@@ -20,6 +20,7 @@ def _write_scope(path: Path, *, confirmed: bool = True, callback_hosts: str = "1
|
||||
domains: [allowed.example]
|
||||
assessment_hosts:
|
||||
callback_hosts: [{callback_hosts}]
|
||||
research_hosts: [github.com, 192.0.2.10]
|
||||
exclusions:
|
||||
ip_addresses: [10.10.10.99]
|
||||
cidrs: [2001:db8:dead::/48]
|
||||
@@ -102,7 +103,7 @@ def test_legacy_descriptive_target_normalises_to_host() -> None:
|
||||
|
||||
def test_callback_hosts_are_secondary_only_and_exclusions_still_win(tmp_path: Path) -> None:
|
||||
scope = tmp_path / "scope.yaml"
|
||||
_write_scope(scope, callback_hosts="10.10.14.5, 10.10.10.99")
|
||||
_write_scope(scope, callback_hosts="10.10.14.5, listener.example, 10.10.10.99")
|
||||
|
||||
callback = check_scope_targets(
|
||||
scope,
|
||||
@@ -112,6 +113,13 @@ def test_callback_hosts_are_secondary_only_and_exclusions_still_win(tmp_path: Pa
|
||||
assert not callback.errors
|
||||
assert not callback.warnings
|
||||
|
||||
for host in ("listener.example", "github.com"):
|
||||
secondary = check_scope_targets(
|
||||
scope, f"curl https://{host}/status", primary_target="10.10.10.10"
|
||||
)
|
||||
assert not secondary.errors
|
||||
assert not secondary.warnings
|
||||
|
||||
unconfigured = check_scope_targets(
|
||||
scope, "bash -c 'echo ready > /dev/tcp/10.10.14.6/4444'", primary_target="10.10.10.10"
|
||||
)
|
||||
@@ -120,7 +128,13 @@ def test_callback_hosts_are_secondary_only_and_exclusions_still_win(tmp_path: Pa
|
||||
callback_as_primary = check_scope_targets(
|
||||
scope, "nc -l -v -s 10.10.14.5 4444", primary_target="10.10.14.5"
|
||||
)
|
||||
assert any("10.10.14.5" in error for error in callback_as_primary.errors)
|
||||
assert any(
|
||||
"secondary-only endpoint 10.10.14.5" in error for error in callback_as_primary.errors
|
||||
)
|
||||
|
||||
for host in ("listener.example", "github.com", "192.0.2.10"):
|
||||
primary = check_scope_targets(scope, f"curl https://{host}", primary_target=host)
|
||||
assert any(f"secondary-only endpoint {host}" in error for error in primary.errors)
|
||||
|
||||
excluded = check_scope_targets(
|
||||
scope, "nc -l -v -s 10.10.10.99 4444", primary_target="10.10.10.10"
|
||||
|
||||
@@ -30,6 +30,9 @@ _SCOPE = """targets:
|
||||
roles:
|
||||
web: 10.10.10.10
|
||||
exclusions: {}
|
||||
assessment_hosts:
|
||||
callback_hosts: [listener.example]
|
||||
research_hosts: [github.com]
|
||||
authorized_parties: ["test owner"]
|
||||
authorisation:
|
||||
confirmed: true
|
||||
@@ -72,6 +75,12 @@ def test_target_role_preserves_ipv6_url_hostname():
|
||||
assert resolve_target(scope, role="web", host_query=None, field="host") == "2001:db8::1"
|
||||
|
||||
|
||||
def test_target_role_preserves_malformed_url_for_review():
|
||||
scope = {"targets": {"roles": {"web": "http://[broken"}}}
|
||||
|
||||
assert resolve_target(scope, role="web", host_query=None, field="host") == "http://[broken"
|
||||
|
||||
|
||||
def _run(*args):
|
||||
return subprocess.run(
|
||||
[sys.executable, str(ROOT / "scripts" / "violin_guard.py"), *args],
|
||||
@@ -141,13 +150,6 @@ def test_target_requires_eng_dir():
|
||||
|
||||
# --- violin_exec_burst -----------------------------------------------------
|
||||
|
||||
_GATE_OK = {
|
||||
"status": "ok",
|
||||
"errors": [],
|
||||
"warnings": [],
|
||||
"infos": [],
|
||||
}
|
||||
|
||||
|
||||
def _patch_burst(monkeypatch, eng_dir):
|
||||
"""Run handle_exec_burst in-process: the real check-command gate is used for
|
||||
@@ -217,6 +219,30 @@ def test_exec_burst_clean_review_or_approved(eng, monkeypatch):
|
||||
assert state.has_pending_sync(str(eng)) is not None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("secondary_only_host", ["listener.example", "github.com"])
|
||||
def test_exec_burst_denies_secondary_only_primary_target(eng, monkeypatch, secondary_only_host):
|
||||
rec = _patch_burst(monkeypatch, str(eng))
|
||||
data = json.loads(
|
||||
service.handle_exec_burst(
|
||||
{
|
||||
"eng_dir": str(eng),
|
||||
"scope": str(eng / "scope" / "scope.yaml"),
|
||||
"phase": "recon",
|
||||
"commands": [f"curl https://{secondary_only_host}"],
|
||||
"target": secondary_only_host,
|
||||
"session_id": "ts",
|
||||
"skill_loaded_file": str(eng / "state" / ".skill-loaded-ts"),
|
||||
"label": "secondary-only-primary",
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
assert data["status"] == "denied"
|
||||
assert data["executed"] == 0
|
||||
assert "secondary-only endpoint" in data["reason"]
|
||||
assert rec["commands"] == []
|
||||
|
||||
|
||||
def test_exec_burst_fail_closed_on_blocked_command(eng, monkeypatch):
|
||||
"""A batch containing a hard-blocked command (e.g. `rm -rf /`) is denied
|
||||
and the batch is halted at the first BLOCK (fail-closed)."""
|
||||
|
||||
@@ -113,6 +113,36 @@ def test_local_source_retrieval_remains_available() -> None:
|
||||
assert result is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"raw_command",
|
||||
[
|
||||
"echo x | nc victim.example 80",
|
||||
"git clone https://github.com/org/repo; curl https://victim.example/admin",
|
||||
"git clone https://github.com/org/repo && nmap victim.example",
|
||||
(
|
||||
"pip install https://files.pythonhosted.org/package.whl "
|
||||
"https://victim.example/package.whl"
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_compound_terminal_commands_cannot_hide_target_segments(raw_command: str) -> None:
|
||||
result = _pre_tool_call_hook(tool_name="terminal", args={"command": raw_command})
|
||||
|
||||
assert result["action"] == "block"
|
||||
assert "violin_exec" in result["message"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"raw_command",
|
||||
[
|
||||
"git clone https://github.com/example/project.git && echo cloned",
|
||||
"echo local | cat",
|
||||
],
|
||||
)
|
||||
def test_safe_compound_terminal_commands_remain_available(raw_command: str) -> None:
|
||||
assert _pre_tool_call_hook(tool_name="terminal", args={"command": raw_command}) is None
|
||||
|
||||
|
||||
def test_safe_local_terminal_command_remains_available() -> None:
|
||||
result = _pre_tool_call_hook(
|
||||
tool_name="terminal",
|
||||
|
||||
@@ -58,7 +58,7 @@ wheels = [
|
||||
|
||||
[[package]]
|
||||
name = "pytest"
|
||||
version = "8.4.2"
|
||||
version = "9.1.1"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
dependencies = [
|
||||
{ name = "colorama", marker = "sys_platform == 'win32'" },
|
||||
@@ -67,9 +67,9 @@ dependencies = [
|
||||
{ name = "pluggy" },
|
||||
{ name = "pygments" },
|
||||
]
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/a3/5c/00a0e072241553e1a7496d638deababa67c5058571567b92a7eaa258397c/pytest-8.4.2.tar.gz", hash = "sha256:86c0d0b93306b961d58d62a4db4879f27fe25513d4b969df351abdddb3c30e01", size = 1519618, upload-time = "2025-09-04T14:34:22.711Z" }
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/e4/47/b9efed96c114afcfa3c9d3fe98a76a1d14c74a9e266d397cf6eb64be5e01/pytest-9.1.1.tar.gz", hash = "sha256:1088fbde8f2b49d95a549a195707afa7a76a3ce9bcadc26b6d71f0ffda5fe313", size = 1636369, upload-time = "2026-06-19T10:58:32.857Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/a8/a4/20da314d277121d6534b3a980b29035dcd51e6744bd79075a6ce8fa4eb8d/pytest-8.4.2-py3-none-any.whl", hash = "sha256:872f880de3fc3a5bdc88a11b39c9710c3497a547cfa9320bc3c5e62fbf272e79", size = 365750, upload-time = "2025-09-04T14:34:20.226Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/24/25/1de2678b631f5a49215c6c96fff41ba892b0a34df68d6d80292b1b48aa7f/pytest-9.1.1-py3-none-any.whl", hash = "sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c", size = 386536, upload-time = "2026-06-19T10:58:31.347Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -154,7 +154,7 @@ wheels = [
|
||||
|
||||
[[package]]
|
||||
name = "violin"
|
||||
version = "2.0.0"
|
||||
version = "2.0.6"
|
||||
source = { virtual = "." }
|
||||
dependencies = [
|
||||
{ name = "filelock" },
|
||||
@@ -172,7 +172,7 @@ requires-dist = [{ name = "filelock", specifier = ">=3.13,<4" }]
|
||||
|
||||
[package.metadata.requires-dev]
|
||||
dev = [
|
||||
{ name = "pytest", specifier = ">=8.0,<9" },
|
||||
{ name = "pytest", specifier = ">=9.0.3,<10" },
|
||||
{ name = "pyyaml", specifier = ">=6.0,<7" },
|
||||
{ name = "ruff", specifier = ">=0.11,<0.12" },
|
||||
]
|
||||
|
||||
Reference in New Issue
Block a user