diff --git a/secator/ai/guardrails.py b/secator/ai/guardrails.py index 37e21029a..5a27c22cf 100644 --- a/secator/ai/guardrails.py +++ b/secator/ai/guardrails.py @@ -525,18 +525,63 @@ class PermissionEngine: Two-step validation: (1) action type check, (2) target/path check. """ - def __init__(self, config: Dict, targets: List[str] = None, workspace: str = ""): + def __init__( + self, config: Dict, targets: List[str] = None, workspace: str = "", + allowed_targets: List[str] = None, denied_targets: List[str] = None + ): self.targets = targets or [] self.workspace = str(workspace) self.rules = {"allow": [], "deny": [], "ask": []} self.runtime_allow: List[Tuple[str, List[str]]] = [] + # Platform-supplied allow-list of target regexes (e.g. validated workspace + # mandates). When set, a `target(...)` action is allowed only if it matches + # one of these regexes — this constrains the AI to the authorized scope. + # Each entry is matched as a regex (full-match), falling back to a literal + # match if the pattern is not valid regex. + self.allowed_targets: List = [] + for pat in (allowed_targets or []): + if not pat: + continue + try: + self.allowed_targets.append(re.compile(pat)) + except re.error: + self.allowed_targets.append(re.compile(re.escape(pat))) + + # Platform-supplied deny-list of target regexes (e.g. the `deny` scope of + # validated workspace mandates). Symmetric to allowed_targets but DENY WINS: + # a `target(...)` matching one of these is denied even if it also matches an + # allowed_targets entry — mirroring the mandate scope matcher's deny-wins. + # Same regex-or-literal compilation as allowed_targets. + self.denied_targets: List = [] + for pat in (denied_targets or []): + if not pat: + continue + try: + self.denied_targets.append(re.compile(pat)) + except re.error: + self.denied_targets.append(re.compile(re.escape(pat))) + for category in ("allow", "deny", "ask"): for rule_str in config.get(category, []): resolved = self._resolve_variables(rule_str) rule_type, patterns = parse_rule(resolved) self.rules[category].append((rule_type, patterns)) + def _matches_allowed_targets(self, value: str) -> bool: + """Check if a target value matches any platform-supplied allowed_targets regex.""" + for rx in self.allowed_targets: + if rx.fullmatch(value) or rx.match(value): + return True + return False + + def _matches_denied_targets(self, value: str) -> bool: + """Check if a target value matches any platform-supplied denied_targets regex.""" + for rx in self.denied_targets: + if rx.fullmatch(value) or rx.match(value): + return True + return False + def _resolve_variables(self, rule: str) -> str: """Replace {workspace} and {targets} variables in a rule string.""" result = rule.replace("{workspace}", self.workspace) @@ -621,6 +666,11 @@ def check_action(self, action: Dict) -> PermissionResult: def _has_rules_for(self, rule_type: str) -> bool: """Check if any rules exist for the given rule type.""" + # Platform-supplied allowed_targets / denied_targets act as a target + # allow/deny-list: their presence forces the target-check step to run so + # out-of-scope targets get constrained and denied targets get blocked. + if rule_type == "target" and (self.allowed_targets or self.denied_targets): + return True for category in ("allow", "deny", "ask"): for rt, _ in self.rules[category]: if rt == rule_type: @@ -701,6 +751,21 @@ def _check_value(self, rule_type: str, value: str) -> PermissionResult: if match_rule(v, patterns): return PermissionResult(decision="deny", reason=f"Denied by rule: {rule_type}({v})") + # Platform-supplied denied_targets (regex) deny-list — checked before the + # allowed_targets allow-list so DENY WINS: a target matching both an allow + # and a deny mandate scope is denied (mirrors the mandate scope matcher). + if rule_type == "target" and self.denied_targets: + for v in values_to_check: + if self._matches_denied_targets(v): + return PermissionResult(decision="deny", reason=f"Denied by mandate: target({v})") + + # Platform-supplied allowed_targets (regex) allow-list — checked after deny + # (deny still wins) but before config/runtime allow rules. + if rule_type == "target" and self.allowed_targets: + for v in values_to_check: + if self._matches_allowed_targets(v): + return PermissionResult(decision="allow", reason=f"Allowed by mandate: target({v})") + for rt, patterns in self.rules["allow"]: if rt == rule_type: for v in values_to_check: diff --git a/secator/tasks/ai.py b/secator/tasks/ai.py index 0f7ca0e2d..00cb4584b 100644 --- a/secator/tasks/ai.py +++ b/secator/tasks/ai.py @@ -63,6 +63,18 @@ class ai(PythonRunner): "internal": True, "help": "Context to pass to AI (findings, scope, objective)" }, + "allowed_targets": { + "type": list, + "default": None, + "internal": True, + "help": "Platform-set allow-list of target strings/regexes the AI must stay within (e.g. validated mandates)" # noqa: E501 + }, + "denied_targets": { + "type": list, + "default": None, + "internal": True, + "help": "Platform-set deny-list of target strings/regexes the AI must never touch (deny wins over allowed_targets)" # noqa: E501 + }, "subagent": { "is_flag": True, "default": False, @@ -429,6 +441,8 @@ def _init_options(self): self.passed_context = self.run_opts.get("context") or {} self.async_tasks = self.get_opt_value("async_tasks") self.dangerous = self.get_opt_value("dangerous") + self.allowed_targets = self.get_opt_value("allowed_targets") or [] + self.denied_targets = self.get_opt_value("denied_targets") or [] # Interactive mode: "local" / "remote" / "auto" interactive = self.get_opt_value("interactive") @@ -452,7 +466,9 @@ def _init_options(self): self.permission_engine = PermissionEngine( CONFIG.addons.ai.permissions, targets=self.inputs, - workspace=self.reports_folder or "" + workspace=self.reports_folder or "", + allowed_targets=self.allowed_targets, + denied_targets=self.denied_targets, ) # Create interactivity backend diff --git a/tests/unit/test_ai_guardrails.py b/tests/unit/test_ai_guardrails.py index a73f9559f..275de28f4 100644 --- a/tests/unit/test_ai_guardrails.py +++ b/tests/unit/test_ai_guardrails.py @@ -341,6 +341,135 @@ def test_default_ask_when_no_rules_match(self): self.assertIn("nmap", result.shell_command) +@unittest.skipUnless(ADDONS_ENABLED['ai'], 'ai addon not installed') +class TestAllowedTargets(unittest.TestCase): + """Platform-supplied allowed_targets (e.g. validated mandates) constrain target scope.""" + + def _make_engine(self, allow=None, deny=None, ask=None, targets=None, allowed_targets=None, workspace="/tmp/workspace"): # noqa: E501 + config = {"allow": allow or [], "deny": deny or [], "ask": ask or []} + return PermissionEngine( + config, targets=targets or [], workspace=workspace, allowed_targets=allowed_targets or []) + + def test_allowed_target_literal_match(self): + engine = self._make_engine(allow=["shell(nmap)"], allowed_targets=["example.com"]) + result = engine.check_action({"action": "shell", "command": "nmap example.com"}) + self.assertEqual(result.decision, "allow") + + def test_allowed_target_regex_match(self): + engine = self._make_engine(allow=["shell(nmap)"], allowed_targets=[r".*\.example\.com"]) + result = engine.check_action({"action": "shell", "command": "nmap api.example.com"}) + self.assertEqual(result.decision, "allow") + + def test_target_outside_allowed_targets_is_constrained(self): + """A target not matching any allowed_targets regex must NOT be silently allowed.""" + engine = self._make_engine(allow=["shell(nmap)"], allowed_targets=[r".*\.example\.com"]) + result = engine.check_action({"action": "shell", "command": "nmap evil.attacker.com"}) + self.assertNotEqual(result.decision, "allow") + + def test_allowed_targets_presence_forces_target_check(self): + """Even with no config target rules, allowed_targets makes the target step run.""" + engine = self._make_engine(allow=["task(*)"], allowed_targets=["10.0.0.1"]) + result = engine.check_action({"action": "task", "name": "nmap", "targets": ["8.8.8.8"]}) + self.assertNotEqual(result.decision, "allow") + + def test_allowed_target_task_in_scope(self): + engine = self._make_engine(allow=["task(*)"], allowed_targets=[r"10\.0\.0\.\d+"]) + result = engine.check_action({"action": "task", "name": "nmap", "targets": ["10.0.0.5"]}) + self.assertEqual(result.decision, "allow") + + def test_deny_still_wins_over_allowed_targets(self): + engine = self._make_engine( + allow=["shell(nmap)"], deny=["target(169.254.169.254)"], allowed_targets=[r".*"]) + result = engine.check_action({"action": "shell", "command": "nmap 169.254.169.254"}) + self.assertEqual(result.decision, "deny") + + def test_invalid_regex_falls_back_to_literal(self): + # '[' is invalid regex → treated as a literal string + engine = self._make_engine(allow=["shell(nmap)"], allowed_targets=["host[1"]) + self.assertEqual(len(engine.allowed_targets), 1) + + def test_allowed_target_url_host_component(self): + engine = self._make_engine(allow=["shell(curl)"], allowed_targets=["example.com"]) + result = engine.check_action({"action": "shell", "command": "curl https://example.com/path"}) + self.assertEqual(result.decision, "allow") + + +@unittest.skipUnless(ADDONS_ENABLED['ai'], 'ai addon not installed') +class TestDeniedTargets(unittest.TestCase): + """Platform-supplied denied_targets (e.g. mandate deny scope) block target scope. Deny wins.""" + + def _make_engine(self, allow=None, deny=None, ask=None, targets=None, allowed_targets=None, # noqa: E501 + denied_targets=None, workspace="/tmp/workspace"): + config = {"allow": allow or [], "deny": deny or [], "ask": ask or []} + return PermissionEngine( + config, targets=targets or [], workspace=workspace, + allowed_targets=allowed_targets or [], denied_targets=denied_targets or []) + + def test_denied_target_literal_match(self): + # IP targets are extracted without DNS, so deny applies directly. + engine = self._make_engine(allow=["shell(nmap)"], denied_targets=["10.0.0.1"]) + result = engine.check_action({"action": "shell", "command": "nmap 10.0.0.1"}) + self.assertEqual(result.decision, "deny") + + def test_denied_target_regex_match(self): + # Hostnames are only extracted if they resolve — patch DNS so the host is seen. + engine = self._make_engine(allow=["shell(nmap)"], denied_targets=[r".*\.evil\.com"]) + with patch("secator.ai.guardrails._resolves", return_value=True): + result = engine.check_action({"action": "shell", "command": "nmap api.evil.com"}) + self.assertEqual(result.decision, "deny") + + def test_deny_wins_over_allow_when_target_matches_both(self): + """A target matching BOTH allowed_targets and denied_targets must be DENIED.""" + engine = self._make_engine( + allow=["shell(nmap)"], allowed_targets=[r".*\.example\.com"], denied_targets=[r"admin\.example\.com"]) + with patch("secator.ai.guardrails._resolves", return_value=True): + result = engine.check_action({"action": "shell", "command": "nmap admin.example.com"}) + self.assertEqual(result.decision, "deny") + + def test_deny_wins_over_allow_at_check_value_level(self): + """Unit-level deny-wins: _check_value denies a target in both allow + deny lists.""" + engine = self._make_engine( + allow=["shell(nmap)"], allowed_targets=[r".*"], denied_targets=[r"169\.254\.169\.254"]) + result = engine._check_value("target", "169.254.169.254") + self.assertEqual(result.decision, "deny") + + def test_allow_only_target_is_allowed(self): + """allow-only (in allowed, not in denied) → allowed.""" + engine = self._make_engine( + allow=["shell(nmap)"], allowed_targets=[r".*"], denied_targets=[r"169\.254\.169\.254"]) + result = engine._check_value("target", "10.0.0.5") + self.assertEqual(result.decision, "allow") + + def test_deny_only_target_is_denied(self): + """deny-only (matches denied, no allowed entries) → denied.""" + engine = self._make_engine(allow=["shell(nmap)"], denied_targets=[r"10\.0\.0\.1"]) + result = engine.check_action({"action": "shell", "command": "nmap 10.0.0.1"}) + self.assertEqual(result.decision, "deny") + + def test_denied_targets_presence_forces_target_check(self): + """Even with no config target rules, denied_targets makes the target step run.""" + engine = self._make_engine(allow=["task(*)"], denied_targets=["10.0.0.1"]) + result = engine.check_action({"action": "task", "name": "nmap", "targets": ["10.0.0.1"]}) + self.assertEqual(result.decision, "deny") + + def test_denied_target_task_in_deny_scope(self): + engine = self._make_engine( + allow=["task(*)"], allowed_targets=[r"10\.0\.0\.\d+"], denied_targets=[r"10\.0\.0\.1"]) + result = engine.check_action({"action": "task", "name": "nmap", "targets": ["10.0.0.1"]}) + self.assertEqual(result.decision, "deny") + + def test_denied_target_url_host_component(self): + engine = self._make_engine(allow=["shell(curl)"], denied_targets=["evil.com"]) + with patch("secator.ai.guardrails._resolves", return_value=True): + result = engine.check_action({"action": "shell", "command": "curl https://evil.com/path"}) + self.assertEqual(result.decision, "deny") + + def test_invalid_regex_falls_back_to_literal(self): + # '[' is invalid regex → treated as a literal string + engine = self._make_engine(allow=["shell(nmap)"], denied_targets=["host[1"]) + self.assertEqual(len(engine.denied_targets), 1) + + @unittest.skipUnless(ADDONS_ENABLED['ai'], 'ai addon not installed') class TestTargetPrompt(unittest.TestCase):