Enforce scope authorization and exclusions

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
 

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:
+    allowed_actions = roe.get("allowed_actions") if isinstance(roe, dict) else None
-        result.add_error("scope.rules_of_engagement.allowed_actions is required")
+    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"(?<![\w:])\[?[0-9A-Fa-f:]{2,}\]?(?:/\d{1,3})?", command):
+        candidate = token.strip("[]").lower()
+        try:
+            ipaddress.ip_network(candidate, strict=False)
+        except ValueError:
+            continue
+        cands.append(candidate)
     for m in _HOST_PORT.finditer(command):
         cands.append(m.group(1).lower())
     for m in _FQDN.finditer(command):
@@ -225,35 +280,63 @@ def _extract_target_candidates(command: str) -> 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 []:
+    for key in ("ip_addresses", "in_scope_urls", "urls", "domains", "hostnames", "roles"):
-        allowed.add(str(ip).lower())
+        for value in _values(targets.get(key, [])):
-    for url in targets.get("in_scope_urls", []) or []:
+            allowed.add(_normalise_scope_host(value))
-        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())
     return allowed
 
 
 def _scope_excluded_hosts(scope: dict) -> set[str]:
     excluded: set[str] = set()
-    for item in scope.get("exclusions", {}) or []:
+    for item in _values(scope.get("exclusions", {})):
-        if isinstance(item, str):
+        excluded.add(_normalise_scope_host(item))
-            excluded.add(item.lower())
-        elif isinstance(item, dict):
-            for v in item.values():
-                excluded.add(str(v).lower())
     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)

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: []

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"

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, (
+    validation = command.validate_scope(eng / "scope" / "scope.yaml")
-        "filled scope must be guard-clean"
+    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):

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

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(