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

secator/ai/guardrails.py

@@ -525,18 +525,39 @@ 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):
 		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)))
+
 		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 _resolve_variables(self, rule: str) -> str:
 		"""Replace {workspace} and {targets} variables in a rule string."""
 		result = rule.replace("{workspace}", self.workspace)
@@ -621,6 +642,10 @@ 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 act as a target allow-list: their presence
+		# forces the target-check step to run so out-of-scope targets get constrained.
+		if rule_type == "target" and self.allowed_targets:
+			return True
 		for category in ("allow", "deny", "ask"):
 			for rt, _ in self.rules[category]:
 				if rt == rule_type:
@@ -701,6 +726,13 @@ 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 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,12 @@ 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
+		},
 		"subagent": {
 			"is_flag": True,
 			"default": False,
@@ -429,6 +435,7 @@ 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 []
 
 		# Interactive mode: "local" / "remote" / "auto"
 		interactive = self.get_opt_value("interactive")
@@ -452,7 +459,8 @@ 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,
 		)
 
 		# Create interactivity backend

tests/unit/test_ai_guardrails.py

@@ -341,6 +341,59 @@ 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 TestTargetPrompt(unittest.TestCase):