feat(ai): constrain ai task to allowed_targets scope (AI-hardening round 1)

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:

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

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):