diff --git a/plugins/violin_guard/__init__.py b/plugins/violin_guard/__init__.py index 796a7a6..7b9a8ae 100644 --- a/plugins/violin_guard/__init__.py +++ b/plugins/violin_guard/__init__.py @@ -217,13 +217,7 @@ def _post_tool_call_hook(tool_name=None, args=None, result=None, duration_ms=0, def _pre_llm_call_hook(session_id=None, eng_dir=None, **kwargs): - """Lifecycle heartbeat: tick the message counter before each LLM call. - - This is the *supplementary* heartbeat — it advances the message cadence - (used to surface periodic review locks) but never replaces the authoritative - command gate. When ``eng_dir`` is available we record the tick; otherwise we - simply return without mutating state. - """ + """Lifecycle heartbeat: tick the message counter before each LLM call.""" if eng_dir: with contextlib.suppress(Exception): state.tick_message(str(eng_dir)) @@ -235,7 +229,7 @@ def _pre_llm_call_hook(session_id=None, eng_dir=None, **kwargs): def _on_session_reset_hook(session_id=None, eng_dir=None, **kwargs) -> None: """Hook: session reset (context compression, /goal set, etc.).""" - eng_dir = eng_dir or _SESSION_ENGAGEMENTS.get(str(session_id or "")) + eng_dir = eng_dir or (_SESSION_ENGAGEMENTS.get(str(session_id)) if session_id else None) if eng_dir: with contextlib.suppress(Exception): state.tick_message(str(eng_dir)) diff --git a/plugins/violin_guard/bootstrap.py b/plugins/violin_guard/bootstrap.py index 925ed2b..990cf42 100644 --- a/plugins/violin_guard/bootstrap.py +++ b/plugins/violin_guard/bootstrap.py @@ -42,6 +42,7 @@ _ARTIFACT_DIRECTORIES = ( "evidence/flags", "evidence/reporting", "evidence/retrospective", + "evidence/executions", ) diff --git a/plugins/violin_guard/command.py b/plugins/violin_guard/command.py index 6dd3f97..7fe2b97 100644 --- a/plugins/violin_guard/command.py +++ b/plugins/violin_guard/command.py @@ -19,8 +19,8 @@ from .results import GuardResult from .skill_receipts import get_binding from .targets import ( check_scope_targets, - extract_target_candidates, normalise_target, + resolve_command_targets, ) __all__ = [ @@ -247,7 +247,11 @@ def check_skill_binding(eng_dir: Path, task_id: str, session_id: str, phase: Pha def check_hypothesis_freshness( - eng_dir: Path, phase: Phase, command: str, primary_target: str | None = None + eng_dir: Path, + phase: Phase, + command: str, + primary_target: str | None = None, + hypothesis_id: str | None = None, ) -> HypothesisResult: """Ensure hypotheses exist and are fresh for phases that require them.""" result = HypothesisResult() @@ -269,9 +273,16 @@ def check_hypothesis_freshness( Phase.PRIVESC: {Phase.EXPLOITATION, Phase.POST_EXPLOITATION, Phase.PRIVESC}, Phase.FLAGS: {Phase.PRIVESC, Phase.FLAGS}, }.get(phase, {phase}) - targets = {normalise_target(target) for target in extract_target_candidates(command)} - if primary_target: - targets.add(normalise_target(primary_target)) + scope_path = eng_dir / "scope" / "scope.yaml" + scope_data = validate_scope(scope_path).scope_data if scope_path.exists() else None + targets = resolve_command_targets(command, primary_target=primary_target, scope_data=scope_data) + + norm_hyp_id = ( + hypothesis_id.strip().upper().removeprefix("H-").lstrip("0") or "0" + if hypothesis_id + else None + ) + relevant = [] for hypothesis in hyps: if hypothesis.canonical_status() == "Rejected" or not hypothesis.target: @@ -281,6 +292,12 @@ def check_hypothesis_freshness( except ValueError: continue target = normalise_target(hypothesis.target) + + if norm_hyp_id is not None: + h_id = hypothesis.id.strip().upper().removeprefix("H-").lstrip("0") or "0" + if h_id != norm_hyp_id: + continue + if hypothesis_phase in acceptable_phases and (not targets or target in targets): relevant.append(hypothesis) if not relevant: @@ -289,11 +306,11 @@ def check_hypothesis_freshness( for h in hyps if h.canonical_status() != "Rejected" and h.target ] - result.add_error( - f"phase {phase.value} requires a non-rejected hypothesis matching the command target; " - f"parsed targets: {', '.join(sorted(targets)) or 'none'}; " - f"available hypotheses: {', '.join(eligible) or 'none'}" - ) + msg = f"phase {phase.value} requires a non-rejected hypothesis matching the command target" + if norm_hyp_id: + msg += f" (linked H-{norm_hyp_id.zfill(3)})" + msg += f"; parsed targets: {', '.join(sorted(targets)) or 'none'}; available hypotheses: {', '.join(eligible) or 'none'}" + result.add_error(msg) return result if phase in { @@ -386,7 +403,7 @@ def check_command(args: CheckCommandArgs) -> CheckResult: artifact_result = check_local_artifact_paths(args.command) result.infos.extend(artifact_result.infos) - # 3. Session identity (legacy markers may infer identity, but never authorize work). + # 3. Session identity gate session_id = state.resolve_session_id(eng_dir, args.session_id) if not session_id: result.add_error("session_id is required for the skill receipt gate") @@ -396,6 +413,7 @@ def check_command(args: CheckCommandArgs) -> CheckResult: ptt_validation = ptt.validate_ptt(ptt.parse_ptt(ptt_path)) result.errors.extend(ptt_validation.errors) result.warnings.extend(ptt_validation.warnings) + active_task_hyp_id = None if ptt_validation.active_task: result.infos.append(f"active PTT task: {ptt_validation.active_task}") active_task = ptt.find_active_task(ptt_validation.tasks) @@ -406,6 +424,10 @@ def check_command(args: CheckCommandArgs) -> CheckResult: "pause the current task and start one under the requested Phase heading with " "violin_record_ptt" ) + if active_task and active_task.note: + hyp_match = re.search(r"\bH-\d+\b", active_task.note, re.IGNORECASE) + if hyp_match: + active_task_hyp_id = hyp_match.group(0).upper() if active_task and session_id: binding_result = check_skill_binding(eng_dir, active_task.id, session_id, phase) result.errors.extend(binding_result.errors) @@ -430,7 +452,9 @@ def check_command(args: CheckCommandArgs) -> CheckResult: result.infos.extend(h_infos) # 6. Hypothesis freshness - hyp_result = check_hypothesis_freshness(eng_dir, phase, args.command, args.target) + hyp_result = check_hypothesis_freshness( + eng_dir, phase, args.command, args.target, hypothesis_id=active_task_hyp_id + ) result.errors.extend(hyp_result.errors) result.warnings.extend(hyp_result.warnings) result.infos.extend(hyp_result.infos) diff --git a/plugins/violin_guard/execution.py b/plugins/violin_guard/execution.py index 5f0b480..428341a 100644 --- a/plugins/violin_guard/execution.py +++ b/plugins/violin_guard/execution.py @@ -155,18 +155,34 @@ def _preview(path: Path) -> str: return handle.read(PREVIEW_BYTES).decode("utf-8", errors="replace") +def _find_execution_manifest(engagement: Path, execution_id: str) -> Path | None: + evidence_dir = engagement / "evidence" / "executions" + if not evidence_dir.exists(): + return None + short_id = execution_id[:8] + candidates = list(evidence_dir.glob(f"*-{short_id}-*.json")) + direct = evidence_dir / f"{execution_id}.json" + if direct.exists() and direct not in candidates: + candidates.append(direct) + for path in candidates: + with state.lock_file(path): + data = state.read_json(path) + if data.get("execution_id") == execution_id: + return path + return None + + def _finalize_background( *, engagement: Path, - registry_path: Path, manifest_path: Path, command: str, phase: str, exit_code: int, status_name: str, ) -> dict[str, Any]: - with state.lock_file(registry_path): - record = state.read_json(registry_path) + with state.lock_file(manifest_path): + record = state.read_json(manifest_path) if record.get("history_recorded"): return record if record.get("cancel_requested"): @@ -190,7 +206,6 @@ def _finalize_background( ) receipt["history_recorded"] = True state.atomic_json(manifest_path, receipt) - state.atomic_json(registry_path, receipt) return receipt @@ -198,7 +213,6 @@ def _monitor_background( proc: subprocess.Popen, *, engagement: Path, - registry_path: Path, manifest_path: Path, stdout_path: Path, stderr_path: Path, @@ -209,7 +223,7 @@ def _monitor_background( deadline = time.monotonic() + timeout status_name = "completed" while proc.poll() is None: - current = state.read_json(registry_path) + current = state.read_json(manifest_path) if current.get("cancel_requested"): status_name = "cancelled" _terminate_process(proc) @@ -231,7 +245,6 @@ def _monitor_background( exit_code = proc.wait(timeout=5) _finalize_background( engagement=engagement, - registry_path=registry_path, manifest_path=manifest_path, command=command, phase=phase, @@ -251,7 +264,6 @@ def _start_background_monitor( *, record: dict[str, Any], engagement: Path, - registry_path: Path, manifest_path: Path, stdout_path: Path, stderr_path: Path, @@ -272,7 +284,6 @@ def _start_background_monitor( kwargs={ "proc": proc, "engagement": engagement, - "registry_path": registry_path, "manifest_path": manifest_path, "stdout_path": stdout_path, "stderr_path": stderr_path, @@ -319,7 +330,6 @@ def execute( stdout_path = evidence_dir / f"{stem}.stdout.txt" stderr_path = evidence_dir / f"{stem}.stderr.txt" manifest_path = evidence_dir / f"{stem}.json" - registry_path = engagement / "state" / "executions" / f"{execution_id}.json" rel_manifest = manifest_path.relative_to(engagement).as_posix() rel_stdout = stdout_path.relative_to(engagement).as_posix() rel_stderr = stderr_path.relative_to(engagement).as_posix() @@ -345,7 +355,7 @@ def execute( "stderr": rel_stderr, }, } - state.atomic_json(registry_path, record) + state.atomic_json(manifest_path, record) timed_out = False output_limited = False @@ -372,14 +382,13 @@ def execute( proc = subprocess.Popen(process_argv, **popen_kwargs) record.update(status="running", pid=proc.pid) - state.atomic_json(registry_path, record) + state.atomic_json(manifest_path, record) if background: return _start_background_monitor( proc, record=record, engagement=engagement, - registry_path=registry_path, manifest_path=manifest_path, stdout_path=stdout_path, stderr_path=stderr_path, @@ -392,7 +401,7 @@ def execute( deadline = time.monotonic() + timeout while proc.poll() is None: - current = state.read_json(registry_path) + current = state.read_json(manifest_path) if current.get("cancel_requested"): cancelled = True _terminate_pid(proc.pid) @@ -435,7 +444,6 @@ def execute( "output_limited": output_limited, } state.atomic_json(manifest_path, receipt) - state.atomic_json(registry_path, receipt) append_history(engagement, command, phase, exit_code, rel_manifest) @@ -471,14 +479,16 @@ def status(eng_dir: str, execution_id: str) -> dict[str, Any]: engagement = _resolve_engagement(eng_dir) if not re.fullmatch(r"[0-9a-fA-F-]{36}", execution_id): raise ValueError("invalid execution_id") - path = engagement / "state" / "executions" / f"{execution_id}.json" + manifest_path = _find_execution_manifest(engagement, execution_id) + if not manifest_path: + raise ValueError("execution not found") # Background finalization replaces this file atomically while status calls # may arrive from another thread. On Windows, reading during the replace # can transiently raise an OSError, which read_json intentionally maps to # an empty document. Serialize the read with the finalizer's lock so a # tracked execution is never misreported as missing. - with state.lock_file(path): - record = state.read_json(path) + with state.lock_file(manifest_path): + record = state.read_json(manifest_path) if not record: raise ValueError("execution not found") if record.get("background") and record.get("status") == "running": @@ -486,8 +496,7 @@ def status(eng_dir: str, execution_id: str) -> dict[str, Any]: if isinstance(pid, int) and pid > 0 and not _pid_is_running(pid): record = _finalize_background( engagement=engagement, - registry_path=path, - manifest_path=engagement / record["evidence_paths"]["manifest"], + manifest_path=manifest_path, command=record["command"], phase=record["phase"], exit_code=-1, @@ -506,8 +515,8 @@ def _pid_is_running(pid: int) -> bool: def cancel(eng_dir: str, execution_id: str) -> dict[str, Any]: engagement = _resolve_engagement(eng_dir) - path = engagement / "state" / "executions" / f"{execution_id}.json" record = status(str(engagement), execution_id) + manifest_path = engagement / record["evidence_paths"]["manifest"] if record.get("status") not in {"starting", "running"}: return {**record, "cancel_requested": False, "message": "execution is not running"} @@ -517,7 +526,7 @@ def cancel(eng_dir: str, execution_id: str) -> dict[str, Any]: record["cancel_requested"] = True record["cancel_requested_at"] = _utc_now() - state.atomic_json(path, record) + state.atomic_json(manifest_path, record) _terminate_pid(pid) return {**record, "message": "cancellation requested for tracked process group"} diff --git a/plugins/violin_guard/handlers/base.py b/plugins/violin_guard/handlers/base.py index d555768..86f833d 100644 --- a/plugins/violin_guard/handlers/base.py +++ b/plugins/violin_guard/handlers/base.py @@ -17,7 +17,7 @@ logger = logging.getLogger(__name__) def _running_background_command(eng_dir: str, command: str) -> bool: """Return True if command is currently running as an acknowledged background process.""" - exec_dir = _eng_path(eng_dir) / "state" / "executions" + exec_dir = _eng_path(eng_dir) / "evidence" / "executions" if not exec_dir.exists(): return False for path in exec_dir.glob("*.json"): diff --git a/plugins/violin_guard/history.py b/plugins/violin_guard/history.py index a5f529f..323d704 100644 --- a/plugins/violin_guard/history.py +++ b/plugins/violin_guard/history.py @@ -10,13 +10,22 @@ from __future__ import annotations from datetime import UTC, datetime from pathlib import Path -from .state import lock_file, resolve_eng_dir +from .state import lock_file, read_json, resolve_eng_dir _COMMAND_MARKER = " | command=" _COMMAND_LENGTH_MARKER = " | command_length=" _RECEIPT_MARKER = " | receipt=" +def normalize_command(command: str) -> str: + """Normalize whitespace and newlines in a command string for reliable history matching.""" + if not command: + return "" + lines = command.replace("\r\n", "\n").split("\n") + cleaned_parts = [part.strip() for part in lines if part.strip()] + return " ".join(cleaned_parts) + + def _history_path(eng_dir: str | Path) -> Path: return resolve_eng_dir(eng_dir) / "state" / "history.md" @@ -32,9 +41,10 @@ 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") + clean_command = normalize_command(command) if "\n" in command else command line = ( - f"- {stamp} | phase={phase} | exit_code={exit_code} | command={command}" - f"{_COMMAND_LENGTH_MARKER}{len(command)}" + f"- {stamp} | phase={phase} | exit_code={exit_code} | command={clean_command}" + f"{_COMMAND_LENGTH_MARKER}{len(clean_command)}" ) if receipt_path: line += f"{_RECEIPT_MARKER}{receipt_path}" @@ -50,8 +60,10 @@ def history_contains(eng_dir: str | Path, command: str) -> bool: hist = _history_path(eng_dir) if not hist.exists(): return False + norm_target = normalize_command(command) for line in hist.read_text(encoding="utf-8").splitlines(): - if _recorded_command(line) == command: + rec = _recorded_command(line) + if rec == command or (rec is not None and normalize_command(rec) == norm_target): return True return False @@ -99,23 +111,28 @@ def check_history_staleness( infos.append("history.md is empty — first command will be recorded") return errors, warnings, infos - # History entries are written as ``... | command=``. Compare + # History entries are written as ``... | command=``. Compare # 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] recorded_command = _recorded_command(last_line) - if recorded_command == command and not allow_pending_repeat: + is_repeat = (recorded_command == command) or ( + recorded_command is not None and normalize_command(recorded_command) == normalize_command(command) + ) + if is_repeat and not allow_pending_repeat: errors.append( f"command appears to be an exact repeat of the last recorded command: {last_line}" ) - elif recorded_command == command: + elif is_repeat: infos.append("exact repeat belongs to the pending batch; allowing reconciliation/retry") return errors, warnings, infos __all__ = [ + "normalize_command", "append_history", "history_contains", "check_history_staleness", ] + diff --git a/plugins/violin_guard/state.py b/plugins/violin_guard/state.py index e4d001a..1f95fbd 100644 --- a/plugins/violin_guard/state.py +++ b/plugins/violin_guard/state.py @@ -14,9 +14,7 @@ from typing import Any from filelock import FileLock -# --------------------------------------------------------------------------- # Constants -# --------------------------------------------------------------------------- DEFAULT_SYNC_CREDIT = 5 COMMAND_INTERVAL = 50 @@ -41,9 +39,7 @@ _SESSION_FILE = "session.json" _SEMANTIC_FILE = "semantic-progress.json" -# --------------------------------------------------------------------------- # Path helpers -# --------------------------------------------------------------------------- def _eng_root() -> Path: @@ -57,9 +53,19 @@ def _eng_root() -> Path: def resolve_eng_dir(eng_dir: str | Path) -> Path: """Resolve an engagement directory path (absolute or relative to profile root).""" + if not str(eng_dir).strip() or str(eng_dir).strip() == ".": + cwd = Path.cwd().resolve() + if (cwd / "scope" / "scope.yaml").exists() or (cwd / "hypotheses.md").exists(): + return cwd + return _eng_root() + path = Path(eng_dir).expanduser() if not path.is_absolute(): - path = _eng_root() / path + profile_candidate = (_eng_root() / path).resolve() + cwd_candidate = (Path.cwd() / path).resolve() + if not profile_candidate.exists() and cwd_candidate.exists(): + return cwd_candidate + return profile_candidate return path.resolve() @@ -84,7 +90,9 @@ def resolve_session_id(eng_dir: str | Path, session_id: str | None = None) -> st def record_session_id(eng_dir: str | Path, session_id: str | None) -> None: if session_id and session_id.strip(): - atomic_json(_state_dir(eng_dir) / _SESSION_FILE, {"session_id": session_id.strip()}) + path = _state_dir(eng_dir) / _SESSION_FILE + with lock_file(path): + atomic_json(path, {"session_id": session_id.strip()}) def _state_dir(eng_dir: str | Path) -> Path: @@ -93,9 +101,7 @@ def _state_dir(eng_dir: str | Path) -> Path: return p -# --------------------------------------------------------------------------- # Storage primitives -# --------------------------------------------------------------------------- @contextmanager @@ -108,12 +114,26 @@ def lock_file(path: Path): def read_json(path: Path) -> dict[str, Any]: - """Read a JSON document, returning an empty dict on error or non-dict root.""" + """Read a JSON document, returning an empty dict on missing file or non-dict root. + + Raises OSError or json.JSONDecodeError on corrupt/locked file reads when the file exists, + preventing mutate_json from overwriting existing state with empty dictionaries. + """ + if not path.exists(): + return {} try: data = json.loads(path.read_text(encoding="utf-8")) return data if isinstance(data, dict) else {} except (OSError, json.JSONDecodeError): - return {} + # On read failure when file exists, attempt up to 3 retries for transient locks + for attempt in range(3): + time.sleep(0.02 * (attempt + 1)) + try: + data = json.loads(path.read_text(encoding="utf-8")) + return data if isinstance(data, dict) else {} + except (OSError, json.JSONDecodeError): + pass + raise def atomic_json(path: Path, data: dict[str, Any]) -> None: @@ -145,9 +165,7 @@ def mutate_json(path: Path, mutation) -> Any: return result -# --------------------------------------------------------------------------- # Local command classification -# --------------------------------------------------------------------------- def is_local_bookkeeping_command(command: str) -> bool: @@ -156,9 +174,7 @@ def is_local_bookkeeping_command(command: str) -> bool: return bool(leading) and leading[0] in LOCAL_TOOLS -# --------------------------------------------------------------------------- # Sync credit / pending sync -# --------------------------------------------------------------------------- def _sync_path(eng_dir: str | Path) -> Path: diff --git a/plugins/violin_guard/targets.py b/plugins/violin_guard/targets.py index 79931c5..ea57b06 100644 --- a/plugins/violin_guard/targets.py +++ b/plugins/violin_guard/targets.py @@ -225,7 +225,6 @@ def resolve_target( if not target_val: return None - # Extract the requested field from a URL if "://" in target_val and field in ("ip", "host"): with contextlib.suppress(ValueError): parsed = urlsplit(target_val) @@ -446,11 +445,32 @@ def _is_ip_network(value: str) -> bool: return True +def resolve_command_targets( + command: str, + primary_target: str | None = None, + scope_data: dict[str, Any] | None = None, +) -> set[str]: + """Extract and normalise candidate targets from command, primary target, or scope fallback.""" + targets = {normalise_target(t) for t in extract_target_candidates(command)} + if primary_target: + targets.add(normalise_target(primary_target)) + + if not targets and isinstance(scope_data, dict): + targets_sec = scope_data.get("targets", {}) + if isinstance(targets_sec, dict): + for t in targets_sec.get("ip_addresses", []) or []: + if isinstance(t, str) and t.strip(): + targets.add(normalise_target(t)) + + return targets + + __all__ = [ "TargetCheckResult", "check_scope_targets", "extract_target_candidates", "normalise_target", + "resolve_command_targets", "resolve_target", "scope_hosts", ] diff --git a/plugins/violin_guard/terminal_policy.py b/plugins/violin_guard/terminal_policy.py index 2e3be0e..33f7161 100644 --- a/plugins/violin_guard/terminal_policy.py +++ b/plugins/violin_guard/terminal_policy.py @@ -19,114 +19,21 @@ import re import shlex from urllib.parse import urlsplit -_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", - } +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, ) @@ -192,17 +99,14 @@ def _word_is_target_literal(word: str) -> bool: if authority.count(":") == 1: authority = authority.split(":", 1)[0] - # IPv4 match if _IPV4_RE.fullmatch(authority): return authority not in {"127.0.0.1", "0.0.0.0"} - # IPv6 match with contextlib.suppress(ValueError): clean_ip = authority.strip("[]") ip_obj = ipaddress.ip_address(clean_ip) return not ip_obj.is_loopback and not ip_obj.is_unspecified - # URL match if "://" in value: try: hostname = urlsplit(value).hostname diff --git a/plugins/violin_guard/terminal_rules.py b/plugins/violin_guard/terminal_rules.py new file mode 100644 index 0000000..2ad8684 --- /dev/null +++ b/plugins/violin_guard/terminal_rules.py @@ -0,0 +1,115 @@ +"""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/tests/guard/guards/test_multiline_history_matching.py b/tests/guard/guards/test_multiline_history_matching.py new file mode 100644 index 0000000..c327de9 --- /dev/null +++ b/tests/guard/guards/test_multiline_history_matching.py @@ -0,0 +1,96 @@ +from pathlib import Path +import json + +from plugins.violin_guard import bootstrap, history, state +from plugins.violin_guard import handlers as service + + +def _engagement(tmp_path: Path) -> Path: + eng = tmp_path / "eng" + bootstrap.init_engagement(eng, host="10.10.10.10") + scope = eng / "scope" / "scope.yaml" + scope.write_text( + scope.read_text(encoding="utf-8").replace("confirmed: false", "confirmed: true"), + encoding="utf-8", + ) + ptt_path = eng / "state" / "ptt.md" + ptt_path.write_text( + ptt_path.read_text(encoding="utf-8").replace("| PT-010 | [ ] |", "| PT-010 | [~] |"), + encoding="utf-8", + ) + return eng + + +def test_normalize_command_collapses_newlines_and_whitespace() -> None: + multiline = "cd /app\n mkdir -p build \n echo 'hello world'\n" + normalized = history.normalize_command(multiline) + assert normalized == "cd /app mkdir -p build echo 'hello world'" + + +def test_append_history_sanitizes_multiline_commands(tmp_path: Path) -> None: + eng = _engagement(tmp_path) + multiline = "cd /var/www\ncurl -s http://example.com\nhead -n 10 index.html" + history.append_history(eng, multiline, "RECON", 0, "evidence/executions/test.json") + + hist_file = eng / "state" / "history.md" + assert hist_file.exists() + record_lines = [line for line in hist_file.read_text(encoding="utf-8").splitlines() if line.startswith("- ")] + assert len(record_lines) == 1 # Formatted as single line in history.md + assert "cd /var/www curl -s http://example.com head -n 10 index.html" in record_lines[0] + + +def test_history_contains_matches_multiline_and_whitespace_variants(tmp_path: Path) -> None: + eng = _engagement(tmp_path) + multiline_cmd = "cd /home/kali\n mkdir -p output\n curl http://eloquia.htb/" + history.append_history(eng, multiline_cmd, "RECON", 0, "evidence/executions/1.json") + + # Match exact multiline input string + assert history.history_contains(eng, multiline_cmd) + + # Match normalized single-line representation + single_line_cmd = "cd /home/kali mkdir -p output curl http://eloquia.htb/" + assert history.history_contains(eng, single_line_cmd) + + # Match string with extra newlines/tabs + variant_cmd = "cd /home/kali\n\tmkdir -p output\n\tcurl http://eloquia.htb/\n" + assert history.history_contains(eng, variant_cmd) + + +def test_history_contains_returns_false_when_not_in_history(tmp_path: Path) -> None: + eng = _engagement(tmp_path) + multiline_cmd = "echo 'start'\ncat /etc/passwd\necho 'done'" + + # Clear history.md + hist_file = eng / "state" / "history.md" + if hist_file.exists(): + hist_file.write_text("# History\n", encoding="utf-8") + + assert not history.history_contains(eng, multiline_cmd) + + +def test_batch_review_succeeds_for_multiline_pending_command(tmp_path: Path) -> None: + eng = _engagement(tmp_path) + multiline_cmd = "cd /app\n curl -i http://eloquia.htb/\n wc -l index.html" + + # Mark pending sync with multiline command + state.mark_pending_sync(eng, multiline_cmd, "RECON", "PT-010") + pending = state.get_pending_sync(eng) + assert pending is not None + + # Record history + history.append_history(eng, multiline_cmd, "RECON", 0, "evidence/executions/test.json") + + # Service batch review must succeed without throwing history validation error + result = json.loads( + service.handle_review_batch( + { + "eng_dir": str(eng), + "id": "PT-010", + "status": "[x]", + "note": "Reviewed multiline batch successfully", + } + ) + ) + assert result.get("status") == "ok" + assert result.get("released") is True + assert state.get_pending_sync(eng) is None diff --git a/tests/guard/integration/test_ctf_workflow_fixes.py b/tests/guard/integration/test_ctf_workflow_fixes.py index 6d5d4ab..fc79109 100644 --- a/tests/guard/integration/test_ctf_workflow_fixes.py +++ b/tests/guard/integration/test_ctf_workflow_fixes.py @@ -142,12 +142,11 @@ def test_batch_review_with_running_background_tunnel(ctf_eng): "stderr": f"evidence/executions/{exec_id}.stderr", }, } - exec_file = ctf_eng / "state" / "executions" / f"{exec_id}.json" - exec_file.parent.mkdir(parents=True, exist_ok=True) - state.atomic_json(ctf_eng / "evidence" / "executions" / f"{exec_id}.json", exec_record) - (ctf_eng / "evidence" / "executions" / f"{exec_id}.stdout").write_text("", encoding="utf-8") - (ctf_eng / "evidence" / "executions" / f"{exec_id}.stderr").write_text("", encoding="utf-8") - state.atomic_json(exec_file, exec_record) + exec_dir = ctf_eng / "evidence" / "executions" + exec_dir.mkdir(parents=True, exist_ok=True) + state.atomic_json(exec_dir / f"{exec_id}.json", exec_record) + (exec_dir / f"{exec_id}.stdout").write_text("", encoding="utf-8") + (exec_dir / f"{exec_id}.stderr").write_text("", encoding="utf-8") res_str = service.handle_review_batch( { diff --git a/tests/guard/integration/test_plugin_guard.py b/tests/guard/integration/test_plugin_guard.py index a46be2f..8c8956d 100644 --- a/tests/guard/integration/test_plugin_guard.py +++ b/tests/guard/integration/test_plugin_guard.py @@ -384,6 +384,84 @@ def test_exploitation_requires_cve_and_exploit_research_attempts(tmp_path): assert not allowed.errors, allowed.errors +def test_hypothesis_enforces_scope_target_fallback(tmp_path): + """Verify hypothesis guard checks scope target when command contains no target string.""" + (tmp_path / "scope").mkdir(parents=True, exist_ok=True) + (tmp_path / "scope" / "scope.yaml").write_text( + "targets:\n ip_addresses:\n - 10.129.47.140\n" + "rules_of_engagement:\n allowed_actions: [RECON, EXPLOITATION]\n" + "engagement:\n name: Test\n" + "authorized_parties: [Tester]\n" + "authorisation:\n confirmed: true\n", + encoding="utf-8", + ) + # Hypothesis is for a DIFFERENT target host (192.168.1.1) + (tmp_path / "hypotheses.md").write_text( + "### H-001: Other host\n" + "- **Target:** 192.168.1.1\n" + "- **Status:** Validated\n" + "- **Phase:** EXPLOITATION\n" + "- **CVE Research:** Done\n" + "- **Exploit Research:** Done\n", + encoding="utf-8", + ) + + # Command has no IP string, but scope target 10.129.47.140 should NOT match 192.168.1.1 hypothesis + result = command.check_hypothesis_freshness( + tmp_path, command.Phase.EXPLOITATION, "python3 exploit.py" + ) + assert result.errors, "Expected error when hypothesis target doesn't match scope target" + assert any( + "requires a non-rejected hypothesis matching the command target" in err + for err in result.errors + ) + + +def test_check_command_enforces_active_task_hypothesis_id(tmp_path): + """Verify check_command validates the specific hypothesis ID linked in active PTT task note.""" + (tmp_path / "scope").mkdir(parents=True, exist_ok=True) + (tmp_path / "scope" / "scope.yaml").write_text( + "targets:\n ip_addresses:\n - 10.129.47.140\n" + "rules_of_engagement:\n allowed_actions: [RECON, EXPLOITATION]\n" + "engagement:\n name: Test\n" + "authorized_parties: [Tester]\n" + "authorisation:\n confirmed: true\n", + encoding="utf-8", + ) + (tmp_path / "state").mkdir(parents=True, exist_ok=True) + (tmp_path / "state" / "ptt.md").write_text( + "## Phase: EXPLOITATION\n\n| PT-001 | [~] | Exploit Task | testing H-002 |\n", + encoding="utf-8", + ) + # H-001 has research, but active task links H-002 which has NO research + (tmp_path / "hypotheses.md").write_text( + "### H-001: First\n" + "- **Target:** 10.129.47.140\n" + "- **Status:** Validated\n" + "- **Phase:** EXPLOITATION\n" + "- **CVE Research:** Done\n" + "- **Exploit Research:** Done\n\n" + "### H-002: Linked Task Hypothesis\n" + "- **Target:** 10.129.47.140\n" + "- **Status:** Candidate\n" + "- **Phase:** EXPLOITATION\n" + "- **CVE Research:** \n" + "- **Exploit Research:** \n", + encoding="utf-8", + ) + + cmd_args = command.CheckCommandArgs( + command="python3 exploit.py 10.129.47.140", + phase="EXPLOITATION", + eng_dir=str(tmp_path), + scope=str(tmp_path / "scope" / "scope.yaml"), + session_id="test-session", + ) + res = command.check_command(cmd_args) + assert res.errors + assert any("H-002 missing CVE Research and Exploit Research" in err for err in res.errors) + + def test_record_ptt_can_start_pristine_task(tmp_path, monkeypatch): from plugins.violin_guard.skill_receipts import SkillViewResult diff --git a/tests/guard/state/test_burst_and_target.py b/tests/guard/state/test_burst_and_target.py index 7e76f66..ff06b7f 100644 --- a/tests/guard/state/test_burst_and_target.py +++ b/tests/guard/state/test_burst_and_target.py @@ -56,6 +56,23 @@ def test_relative_engagement_paths_stay_under_profile_root(monkeypatch): assert state.resolve_eng_dir("engagements/demo") == (ROOT / "engagements" / "demo").resolve() +def test_resolve_eng_dir_cwd_and_init_engagement_artifact_dirs(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + (tmp_path / "hypotheses.md").write_text("# Hypotheses\n", encoding="utf-8") + + # Empty string resolves to CWD when engagement markers are present + assert state.resolve_eng_dir("") == tmp_path.resolve() + # Path("") should behave identically to "" + assert state.resolve_eng_dir(Path("")) == tmp_path.resolve() + + bootstrap.init_engagement(tmp_path, host="127.0.0.1") + assert (tmp_path / "evidence" / "executions").is_dir() + + # Relative paths still prefer profile root when neither candidate exists + monkeypatch.delenv("VIOLIN_ENG_ROOT", raising=False) + assert state.resolve_eng_dir("engagements/demo") == (ROOT / "engagements" / "demo").resolve() + + def test_public_handlers_serialize_expected_errors(tmp_path): cases = ( (service.handle_status, {}), diff --git a/tests/guard/test_core_guard_fixes.py b/tests/guard/test_core_guard_fixes.py new file mode 100644 index 0000000..479e29d --- /dev/null +++ b/tests/guard/test_core_guard_fixes.py @@ -0,0 +1,51 @@ +"""Unit tests verifying core guard bug fixes and state hardening.""" + +from plugins.violin_guard import _on_session_reset_hook, command, state +from plugins.violin_guard.phases import Phase + + +def test_hypothesis_zero_parsing(tmp_path): + """Verify hypothesis H-0 or '0' parses as '0' instead of being stripped to empty string.""" + hyp_file = tmp_path / "hypotheses.md" + hyp_file.write_text( + "# Hypotheses\n\n" + "### H-0\n" + "- Status: Formulated\n" + "- Phase: VULN_RESEARCH\n" + "- Target: 10.0.0.1\n" + "- CVE Research: N/A\n" + "- Exploit Research: N/A\n", + encoding="utf-8", + ) + + res = command.check_hypothesis_freshness( + eng_dir=tmp_path, + phase=Phase.VULN_RESEARCH, + command="nmap 10.0.0.1", + primary_target="10.0.0.1", + hypothesis_id="H-0", + ) + assert not any("unlinked" in err.lower() for err in res.errors) + + +def test_read_json_non_existent_vs_error(tmp_path): + """Verify read_json returns {} for non-existent file but raises on persistent read errors.""" + non_existent = tmp_path / "missing.json" + assert state.read_json(non_existent) == {} + + existing = tmp_path / "existing.json" + existing.write_text('{"key": "value"}', encoding="utf-8") + assert state.read_json(existing) == {"key": "value"} + + +def test_resolve_eng_dir_defaults_to_cwd(tmp_path, monkeypatch): + """Verify resolve_eng_dir resolves to CWD when scope markers are present.""" + monkeypatch.chdir(tmp_path) + (tmp_path / "hypotheses.md").write_text("# Hypotheses\n", encoding="utf-8") + assert state.resolve_eng_dir("") == tmp_path.resolve() + assert state.resolve_eng_dir(".") == tmp_path.resolve() + + +def test_on_session_reset_hook_none_session_id(): + """Verify _on_session_reset_hook handles None session_id without throwing or raising KeyError.""" + _on_session_reset_hook(session_id=None, eng_dir=None) diff --git a/tests/guard/test_terminal_policy.py b/tests/guard/test_terminal_policy.py index db4fd7a..66ea4f3 100644 --- a/tests/guard/test_terminal_policy.py +++ b/tests/guard/test_terminal_policy.py @@ -533,4 +533,3 @@ def test_local_script_syntax_and_test_checks_are_allowed(raw_command: str) -> No ) def test_expanded_local_file_tools_are_allowed(raw_command: str) -> None: assert _pre_tool_call_hook(tool_name="terminal", args={"command": raw_command}) is None -