diff --git a/plugins/violin_guard/core/bootstrap.py b/plugins/violin_guard/core/bootstrap.py index b748d67..ed6659f 100644 --- a/plugins/violin_guard/core/bootstrap.py +++ b/plugins/violin_guard/core/bootstrap.py @@ -195,7 +195,12 @@ def init_engagement( result.print() return 1 - result.add_info(f"engagement initialised and guard-clean: {eng_dir}") + if ctf: + result.add_info(f"engagement initialised and ready for authorised CTF work: {eng_dir}") + else: + result.add_info( + f"engagement initialised; confirm scope authorisation before target work: {eng_dir}" + ) result.print() return 0 diff --git a/plugins/violin_guard/core/command.py b/plugins/violin_guard/core/command.py index bf45a06..29a0513 100644 --- a/plugins/violin_guard/core/command.py +++ b/plugins/violin_guard/core/command.py @@ -6,6 +6,7 @@ No subprocess calls — pure functions returning dataclasses. from __future__ import annotations +import ipaddress import re from dataclasses import dataclass, field from datetime import UTC, datetime @@ -24,6 +25,7 @@ __all__ = [ "SkillLoadResult", "check_command", "validate_scope", + "check_scope_authorization", "check_skill_load", "check_history_staleness", "check_hypothesis_freshness", @@ -128,6 +130,14 @@ def validate_scope(scope_path: Path) -> ScopeResult: if section not in data: result.add_error(f"scope.yaml missing required section: {section}") + # A real scope must name the approving party and be explicitly confirmed. + parties = data.get("authorized_parties") + if not isinstance(parties, list) or not any(str(item).strip() for item in parties): + result.add_error("scope.authorized_parties must be a non-empty list") + authorisation = data.get("authorisation") + if not isinstance(authorisation, dict) or authorisation.get("confirmed") is not True: + result.add_error("scope.authorisation.confirmed must be true before target execution") + # targets.ip_addresses targets = data.get("targets", {}) if "ip_addresses" not in targets: @@ -137,8 +147,11 @@ def validate_scope(scope_path: Path) -> ScopeResult: # rules_of_engagement roe = data.get("rules_of_engagement", {}) - if "allowed_actions" not in roe: - result.add_error("scope.rules_of_engagement.allowed_actions is required") + allowed_actions = roe.get("allowed_actions") if isinstance(roe, dict) else None + if not isinstance(allowed_actions, list) or not any( + str(item).strip() for item in allowed_actions + ): + result.add_error("scope.rules_of_engagement.allowed_actions must be a non-empty list") # engagement.date engagement = data.get("engagement", {}) @@ -149,6 +162,39 @@ def validate_scope(scope_path: Path) -> ScopeResult: return result +_PHASE_ACTION_TERMS = { + Phase.SCOPING: ("scope",), + Phase.RECON: ("recon", "discovery", "banner", "version", "scan", "enumerat"), + Phase.VULN_RESEARCH: ("vuln", "research", "cve", "exploitdb"), + Phase.EXPLOITATION: ("exploit", "validation", "poc"), + Phase.POST_EXPLOITATION: ("post-exploit", "post exploitation", "exploit", "validation"), + Phase.PRIVESC: ("privilege", "privesc", "exploit", "validation"), + Phase.FLAGS: ("flag", "capture"), + Phase.REPORTING: ("report",), + Phase.RETROSPECTIVE: ("retrospective",), +} + + +def check_scope_authorization(scope: dict[str, Any] | None, phase: Phase) -> CheckResult: + """Ensure the approved rules of engagement allow the requested phase.""" + result = CheckResult() + if not isinstance(scope, dict): + return result + roe = scope.get("rules_of_engagement") or {} + allowed = [str(item).lower() for item in roe.get("allowed_actions", []) or []] + forbidden = [str(item).lower() for item in roe.get("forbidden_actions", []) or []] + terms = _PHASE_ACTION_TERMS[phase] + if any(any(term in action for term in terms) for action in forbidden): + result.add_error( + f"phase {phase.value} conflicts with scope.rules_of_engagement.forbidden_actions" + ) + if not any(any(term in action for term in terms) for action in allowed): + result.add_error( + f"phase {phase.value} is not permitted by scope.rules_of_engagement.allowed_actions" + ) + return result + + # --------------------------------------------------------------------------- # # DANGEROUS-PATTERN ENFORCEMENT (audit P0: destructive commands were never # blocked). These patterns are hard BLOCKs — yolo cannot bypass them. @@ -212,6 +258,15 @@ def _extract_target_candidates(command: str) -> list[str]: cands.append(host.lower()) for m in _IPV4_CIDR.finditer(command): cands.append(m.group(0).lower()) + # Validate colon-containing tokens with ipaddress instead of treating the + # leading hextet of an IPv6 address as a host:port pair. + for token in re.findall(r"(? list[str]: return out +def _values(value: Any): + if isinstance(value, dict): + for nested in value.values(): + yield from _values(nested) + elif isinstance(value, list): + for nested in value: + yield from _values(nested) + elif value is not None: + yield str(value) + + +def _normalise_scope_host(value: str) -> str: + match = re.match(r"https?://([^\s/]+)", value, flags=re.IGNORECASE) + host = match.group(1) if match else value + if host.startswith("[") and "]" in host: + return host[1 : host.index("]")].lower() + return host.rsplit(":", 1)[0].lower() if host.count(":") == 1 else host.lower() + + def _scope_allowed_hosts(scope: dict) -> set[str]: allowed: set[str] = set() targets = scope.get("targets", {}) or {} - for ip in targets.get("ip_addresses", []) or []: - allowed.add(str(ip).lower()) - for url in targets.get("in_scope_urls", []) or []: - m = re.match(r"https?://([^\s/]+)", str(url)) - if m: - allowed.add(m.group(1).lower()) - roles = targets.get("roles", {}) or {} - if isinstance(roles, dict): - for v in roles.values(): - allowed.add(str(v).lower()) - for h in targets.get("hostnames", []) or []: - allowed.add(str(h).lower()) + for key in ("ip_addresses", "in_scope_urls", "urls", "domains", "hostnames", "roles"): + for value in _values(targets.get(key, [])): + allowed.add(_normalise_scope_host(value)) return allowed def _scope_excluded_hosts(scope: dict) -> set[str]: excluded: set[str] = set() - for item in scope.get("exclusions", {}) or []: - if isinstance(item, str): - excluded.add(item.lower()) - elif isinstance(item, dict): - for v in item.values(): - excluded.add(str(v).lower()) + for item in _values(scope.get("exclusions", {})): + excluded.add(_normalise_scope_host(item)) return excluded +def _scope_networks(scope: dict, section: str) -> list[ipaddress._BaseNetwork]: + values = [] + targets = scope.get(section, {}) or {} + for key in ("ip_addresses", "cidrs"): + values.extend(_values(targets.get(key, []))) + networks = [] + for value in values: + try: + networks.append(ipaddress.ip_network(value, strict=False)) + except ValueError: + continue + return networks + + +def _matches_network(candidate: str, networks: list[ipaddress._BaseNetwork]) -> bool: + try: + network = ipaddress.ip_network(candidate, strict=False) + except ValueError: + return False + return any(network.version == allowed.version and network.subnet_of(allowed) for allowed in networks) + + def check_scope_targets(scope_path: Path, command: str) -> CheckResult: """Block commands whose IP/CIDR target is outside the engagement scope.""" result = CheckResult() @@ -270,12 +353,22 @@ def check_scope_targets(scope_path: Path, command: str) -> CheckResult: allowed = _scope_allowed_hosts(data) excluded = _scope_excluded_hosts(data) + allowed_networks = _scope_networks(data, "targets") + excluded_networks = _scope_networks(data, "exclusions") for cand in _extract_target_candidates(command): - if cand in excluded: + if cand in excluded or _matches_network(cand, excluded_networks): + result.add_error(f"excluded target {cand} must not be touched") continue if cand in allowed: continue - if _IPV4_CIDR.fullmatch(cand): + if _matches_network(cand, allowed_networks): + continue + try: + ipaddress.ip_network(cand, strict=False) + is_ip = True + except ValueError: + is_ip = False + if is_ip: result.add_error(f"out-of-scope target {cand} (not present in scope.yaml)") else: result.add_warning(f"host {cand} is not present in scope.yaml; verify authorization") @@ -406,6 +499,9 @@ def check_command(args: CheckCommandArgs) -> CheckResult: result.errors.extend(scope_result.errors) result.warnings.extend(scope_result.warnings) + authorisation_result = check_scope_authorization(scope_result.scope_data, phase) + result.errors.extend(authorisation_result.errors) + # 2b. Scope target enforcement (audit P0). Extract command targets and # block anything that lands on an out-of-scope IP/CIDR. target_result = check_scope_targets(scope_path, args.command) diff --git a/tests/guard/test_burst_and_target.py b/tests/guard/test_burst_and_target.py index c9585ef..4037d03 100644 --- a/tests/guard/test_burst_and_target.py +++ b/tests/guard/test_burst_and_target.py @@ -23,6 +23,9 @@ _SCOPE = """targets: roles: web: 10.10.10.10 exclusions: {} +authorized_parties: ["test owner"] +authorisation: + confirmed: true rules_of_engagement: allowed_actions: [recon, vuln-research, exploitation] forbidden_actions: [] diff --git a/tests/guard/test_correctness_roadmap_1_1_1.py b/tests/guard/test_correctness_roadmap_1_1_1.py index 4e39d4a..2c58f13 100644 --- a/tests/guard/test_correctness_roadmap_1_1_1.py +++ b/tests/guard/test_correctness_roadmap_1_1_1.py @@ -53,6 +53,9 @@ def _init_e2e(tmp_path, skill_file, allowed=("recon", "vuln-research", "exploita " ip_addresses: [10.10.10.10]\n" " in_scope_urls: []\n" "exclusions: {}\n" + "authorized_parties: [test-owner]\n" + "authorisation:\n" + " confirmed: true\n" "rules_of_engagement:\n" f" allowed_actions: [{', '.join(allowed)}]\n" " forbidden_actions: []\n" diff --git a/tests/guard/test_plugin_guard.py b/tests/guard/test_plugin_guard.py index d2f0a4c..a1699df 100644 --- a/tests/guard/test_plugin_guard.py +++ b/tests/guard/test_plugin_guard.py @@ -19,6 +19,9 @@ _PLATFORM_SCOPE = """targets: ip_addresses: ["10.10.10.10"] in_scope_urls: [] exclusions: {} +authorized_parties: ["test owner"] +authorisation: + confirmed: true rules_of_engagement: allowed_actions: [recon, vuln-research, exploitation] forbidden_actions: [] @@ -323,12 +326,12 @@ def test_init_engagement_creates_compliant_artifacts(tmp_path): rc = bootstrap.init_engagement(str(eng)) assert rc == 0, "init-engagement should succeed" - # scope.yaml present and parses clean (no REVIEW on required fields) + # A default engagement is structurally complete but deliberately remains + # unapproved until the operator confirms authorisation. scope = yaml.safe_load((eng / "scope" / "scope.yaml").read_text(encoding="utf-8")) assert scope["targets"]["ip_addresses"] == ["10.129.45.228"] - assert command.validate_scope(eng / "scope" / "scope.yaml").exit_code() == 0, ( - "filled scope must be guard-clean" - ) + validation = command.validate_scope(eng / "scope" / "scope.yaml") + assert any("authorisation.confirmed" in error for error in validation.errors) # bootstrap reports complete (exit 0) or REVIEW-only (pristine PTT is # legitimate on a brand-new engagement — no task touched yet). @@ -348,11 +351,11 @@ def test_auto_repair_creates_missing_artifacts(tmp_path): # After self-heal, bootstrap must be clean (0) or REVIEW-only (2). assert int(res) in (0, 2), f"auto-repair should self-heal to clean, got {res}" - # Artifacts now exist and scope is guard-clean + # Artifacts now exist; a real operator still has to confirm authorisation. for rel in ("scope/scope.yaml", "state/ptt.md", "hypotheses.md", "state/history.md"): assert (eng / rel).exists(), f"auto-repair should create {rel}" yaml.safe_load((eng / "scope" / "scope.yaml").read_text(encoding="utf-8")) - assert command.validate_scope(eng / "scope" / "scope.yaml").exit_code() == 0 + assert command.validate_scope(eng / "scope" / "scope.yaml").errors def test_exec_auto_records_history_but_requires_explicit_ptt_review(monkeypatch, tmp_path): diff --git a/tests/guard/test_scope_authorization.py b/tests/guard/test_scope_authorization.py new file mode 100644 index 0000000..42dc18f --- /dev/null +++ b/tests/guard/test_scope_authorization.py @@ -0,0 +1,46 @@ +"""Fail-closed authorization and target-scope regression tests.""" + +from __future__ import annotations + +from pathlib import Path + +from plugins.violin_guard.core.command import check_scope_targets, validate_scope + + +def _write_scope(path: Path, *, confirmed: bool = True) -> None: + path.write_text( + f"""targets: + ip_addresses: [10.10.10.10] + cidrs: [2001:db8::/32] + domains: [allowed.example] +exclusions: + ip_addresses: [10.10.10.99] + cidrs: [2001:db8:dead::/48] + domains: [excluded.example] +authorized_parties: [test owner] +authorisation: + confirmed: {str(confirmed).lower()} +rules_of_engagement: + allowed_actions: [host/port discovery, exploit validation] + forbidden_actions: [post-exploitation] +engagement: + date: "2026-07-13" +""", + encoding="utf-8", + ) + + +def test_unconfirmed_scope_is_a_hard_block(tmp_path: Path) -> None: + scope = tmp_path / "scope.yaml" + _write_scope(scope, confirmed=False) + assert any("authorisation.confirmed" in error for error in validate_scope(scope).errors) + + +def test_exclusions_and_ipv6_cidrs_are_enforced(tmp_path: Path) -> None: + scope = tmp_path / "scope.yaml" + _write_scope(scope) + + assert check_scope_targets(scope, "nmap 10.10.10.99").errors + assert check_scope_targets(scope, "nmap 2001:db8:dead::1").errors + assert not check_scope_targets(scope, "nmap 2001:db8:beef::1").errors + assert check_scope_targets(scope, "curl https://excluded.example").errors diff --git a/tests/guard/test_sync_credit_window.py b/tests/guard/test_sync_credit_window.py index b93b26d..2845014 100644 --- a/tests/guard/test_sync_credit_window.py +++ b/tests/guard/test_sync_credit_window.py @@ -11,6 +11,11 @@ from plugins.violin_guard.core import bootstrap, execution, service, state def _engagement(tmp_path: Path) -> Path: eng = tmp_path / "10.10.10.10-2026-07-13" assert bootstrap.init_engagement(eng, host="10.10.10.10") == 0 + scope_path = eng / "scope" / "scope.yaml" + scope_path.write_text( + scope_path.read_text(encoding="utf-8").replace("confirmed: false", "confirmed: true"), + encoding="utf-8", + ) (eng / "state" / ".skill-loaded-test").write_text("skill-loaded: test\n", encoding="utf-8") ptt_path = eng / "state" / "ptt.md" ptt_path.write_text(