.NET: Harden LocalCodeAct provenance and process capability checks

dotnet/src/Microsoft.Agents.AI.LocalCodeAct/README.md

@@ -40,7 +40,8 @@ using var provider = new LocalCodeActProvider("/usr/bin/python3", options);
 ## What the Package Controls
 
 - **AST validation** (default on): Validates generated code against allow-lists
-  before execution.
+  before execution. OS-derived aliases and containers are collected before
+  their uses are checked, including references in helpers and later loop iterations.
 - **Subprocess execution**: Runs generated code in a child Python process.
 - **Explicit Python path**: the provider and standalone function constructors require a Python executable path (no default).
 - **Isolated environment**: Does not inherit host environment variables unless
@@ -57,6 +58,14 @@ validator blocks common dangerous operations (`eval`, `exec`,
 `import subprocess`, attribute access for `os.system`, `__class__`, etc.) but
 does not make Python execution safe on an unsandboxed host.
 
+OS provenance collection uses a finite set of permitted chains and a forbidden
+state so cyclic aliases cannot grow the analysis indefinitely. A dependency
+worklist revisits only assignments that depend on newly discovered provenance.
+Tracking is conservative across scopes and assignments; reusing a name for
+unrelated values can cause validation to reject otherwise benign code. Capability restrictions
+also apply to private asyncio process transports and Windows process helpers,
+including aliases and from-imports.
+
 ## What the Package Does NOT Protect
 
 - Malicious Python code working within allowed imports and operations.

dotnet/src/Microsoft.Agents.AI.LocalCodeAct/Resources/validator.py

@@ -6,6 +6,7 @@
 
 import ast
 import builtins as _builtins
+from collections import deque
 from typing import Any
 
 _PYTHON_BUILTIN_NAMES: frozenset[str] = frozenset(dir(_builtins))
@@ -116,10 +117,15 @@
 
 _BLOCKED_CAPABILITY_ATTRS: frozenset[str] = frozenset(
     {
+        "Popen",
+        "_UnixSubprocessTransport",
+        "_WindowsSubprocessTransport",
         "__builtins__",
         "_eval_type",
         "_evaluate",
+        "_make_subprocess_transport",
         "_sys",
+        "_winapi",
         "ag_frame",
         "builtins",
         "connect_accepted_socket",
@@ -172,6 +178,7 @@
 _OS_ROOT_CHAIN: tuple[str, ...] = ("os",)
 _OS_PATH_CHAIN: tuple[str, ...] = ("os", "path")
 _OS_ENVIRON_CHAIN: tuple[str, ...] = ("os", "environ")
+_OS_FORBIDDEN_CHAIN: tuple[str, ...] = ("<forbidden OS-derived value>",)
 
 # Allowed builtin function names that generated code may call.
 # Note: getattr/setattr/hasattr/delattr are NOT included because they can bypass
@@ -389,6 +396,8 @@ def __init__(
         self._allowed_os_attrs = allowed_os_attrs if allowed_os_attrs is not None else ALLOWED_OS_ATTRS
         self._os_aliases: dict[str, set[tuple[str, ...]]] = {"os": {_OS_ROOT_CHAIN}}
         self._os_containers: dict[str, set[tuple[str, ...]]] = {}
+        self._collecting_provenance = False
+        self._changed_os_names: set[str] = set()
         self._shadowed_os_value_builtins: set[str] = set()
 
     def validate(self, code: str) -> None:
@@ -402,13 +411,80 @@ def validate(self, code: str) -> None:
         self._os_aliases = {"os": {_OS_ROOT_CHAIN}}
         self._os_containers = {}
         self._shadowed_os_value_builtins = self._find_shadowed_os_value_builtins(tree)
+        self._collect_os_provenance(tree)
         self.visit(tree)
 
         if self._errors:
             raise CodeValidationError(
                 "Generated code violates allow-list policy:\n" + "\n".join(f"- {err}" for err in self._errors)
             )
 
+    def _collect_os_provenance(self, tree: ast.AST) -> None:
+        """Collect finite, conservative provenance before checking any uses."""
+        nodes = list(ast.walk(tree))
+        producers: list[ast.AST] = []
+        dependents: dict[str, list[int]] = {}
+        for node in nodes:
+            if isinstance(node, ast.Import):
+                for alias in node.names:
+                    if alias.name == "os" or alias.name.startswith("os."):
+                        chain = tuple(alias.name.split(".")) if alias.asname else _OS_ROOT_CHAIN
+                        self._remember_os_alias(alias.asname or "os", chain)
+            elif isinstance(node, ast.ImportFrom) and node.module is not None:
+                for alias in node.names:
+                    if node.module == "os" or node.module.startswith("os."):
+                        chain = (*tuple(node.module.split(".")), alias.name)
+                        self._remember_os_alias(alias.asname or alias.name, chain)
+                    elif alias.name in {"os", "_os"}:
+                        self._remember_os_alias(alias.asname or alias.name, _OS_ROOT_CHAIN)
+            elif isinstance(node, (ast.Assign, ast.AnnAssign, ast.For, ast.AsyncFor)):
+                value = node.iter if isinstance(node, (ast.For, ast.AsyncFor)) else node.value
+                if value is None:
+                    continue
+                index = len(producers)
+                producers.append(node)
+                for name in {child.id for child in ast.walk(value) if isinstance(child, ast.Name)}:
+                    dependents.setdefault(name, []).append(index)
+
+        chain_count = 2 + len(self._allowed_os_attrs)  # Root, forbidden, and configured top-level members.
+        if "path" in self._allowed_os_attrs:
+            chain_count += len(ALLOWED_OS_PATH_ATTRS)
+        if "environ" in self._allowed_os_attrs:
+            chain_count += len(ALLOWED_OS_ENVIRON_ATTRS)
+
+        pending = deque(range(len(producers)))
+        scheduled = set(pending)
+        # Each direct/contained name fact can schedule each dependent at most once.
+        remaining_propagations = len(producers) + 2 * chain_count * sum(len(items) for items in dependents.values())
+        self._collecting_provenance = True
+        try:
+            while pending:
+                if remaining_propagations == 0:
+                    raise CodeValidationError(
+                        "Generated code is not allowed: OS provenance analysis exceeded its finite bound"
+                    )
+                remaining_propagations -= 1
+                index = pending.popleft()
+                scheduled.remove(index)
+                self._changed_os_names.clear()
+                node = producers[index]
+                if isinstance(node, ast.Assign):
+                    for target in node.targets:
+                        self._track_os_provenance(target, node.value)
+                elif isinstance(node, ast.AnnAssign) and node.value is not None:
+                    self._track_os_provenance(node.target, node.value)
+                elif isinstance(node, (ast.For, ast.AsyncFor)):
+                    self._track_os_iteration_target(node.target, node.iter)
+
+                for name in self._changed_os_names:
+                    for dependent in dependents.get(name, []):
+                        if dependent not in scheduled:
+                            pending.append(dependent)
+                            scheduled.add(dependent)
+        finally:
+            self._collecting_provenance = False
+            self._changed_os_names.clear()
+
     def visit(self, node: ast.AST) -> Any:
         """Visit a node and check if its type is allowed."""
         node_type = type(node)
@@ -426,16 +502,8 @@ def visit_Import(self, node: ast.Import) -> None:
             elif module_name not in self._allowed_imports:
                 self._errors.append(f"Import of '{alias_node.name}' is not allowed (not in allow-list)")
 
-            if alias_node.name == "os":
-                self._remember_os_alias(alias_node.asname or "os", _OS_ROOT_CHAIN)
-            elif alias_node.name.startswith("os."):
-                chain = tuple(alias_node.name.split("."))
-                if chain != _OS_PATH_CHAIN:
-                    self._errors.append(f"Import of '{alias_node.name}' is not allowed")
-                elif alias_node.asname is not None:
-                    self._remember_os_alias(alias_node.asname, chain)
-                else:
-                    self._remember_os_alias("os", _OS_ROOT_CHAIN)
+            if alias_node.name.startswith("os.") and alias_node.name != "os.path":
+                self._errors.append(f"Import of '{alias_node.name}' is not allowed")
         self.generic_visit(node)
 
     def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
@@ -454,19 +522,14 @@ def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
             self._errors.append(f"Import from '{node.module}' is not allowed (not in allow-list)")
         elif module_name == "os":
             for alias_node in node.names:
-                chain = (*tuple(node.module.split(".")), alias_node.name)
                 if node.module != "os" or alias_node.name not in self._allowed_os_attrs:
                     self._errors.append(f"Import from 'os' of '{alias_node.name}' is not allowed")
-                else:
-                    self._remember_os_alias(alias_node.asname or alias_node.name, chain)
         else:
             for alias_node in node.names:
                 if alias_node.name.startswith("__") and alias_node.name.endswith("__"):
                     self._errors.append(
                         f"Import from '{node.module}' of reflective attribute '{alias_node.name}' is not allowed"
                     )
-                elif alias_node.name in {"os", "_os"}:
-                    self._remember_os_alias(alias_node.asname or alias_node.name, _OS_ROOT_CHAIN)
                 elif alias_node.name in _BLOCKED_CAPABILITY_ATTRS:
                     self._errors.append(
                         f"Import from '{node.module}' of capability '{alias_node.name}' is not allowed"
@@ -570,7 +633,10 @@ def visit_Lambda(self, node: ast.Lambda) -> None:
         self.generic_visit(node)
 
     def _remember_os_alias(self, name: str, chain: tuple[str, ...]) -> None:
-        self._os_aliases.setdefault(name, set()).add(chain)
+        self._os_aliases.setdefault(name, set()).add(self._normalize_os_chain(chain))
+
+    def _normalize_os_chain(self, chain: tuple[str, ...]) -> tuple[str, ...]:
+        return chain if self._is_allowed_os_chain(chain) else _OS_FORBIDDEN_CHAIN
 
     def _remember_os_chains(
         self,
@@ -582,15 +648,27 @@ def _remember_os_chains(
             target = target.value
 
         if isinstance(target, ast.Name):
+            if not self._collecting_provenance:
+                return
+            previous_count = len(self._os_aliases.get(target.id, set())) + len(
+                self._os_containers.get(target.id, set())
+            )
             if direct:
-                self._os_aliases.setdefault(target.id, set()).update(direct)
+                self._os_aliases.setdefault(target.id, set()).update(self._normalize_os_chain(chain) for chain in direct)
             if contained:
-                self._os_containers.setdefault(target.id, set()).update(contained)
+                self._os_containers.setdefault(target.id, set()).update(
+                    self._normalize_os_chain(chain) for chain in contained
+                )
+            fact_count = len(self._os_aliases.get(target.id, set())) + len(
+                self._os_containers.get(target.id, set())
+            )
+            if fact_count > previous_count:
+                self._changed_os_names.add(target.id)
         elif isinstance(target, (ast.Tuple, ast.List)):
             possible = direct | contained
             for target_item in target.elts:
                 self._remember_os_chains(target_item, possible, set())
-        elif direct or contained:
+        elif not self._collecting_provenance and (direct or contained):
             self._errors.append("Storing an OS-derived value on an object is not allowed")
 
     def _track_os_provenance(self, target: ast.AST, value: ast.AST) -> None:
@@ -660,9 +738,10 @@ def _get_os_provenance(
 
         if isinstance(node, ast.Attribute):
             direct, _ = self._get_os_provenance(node.value)
-            if not direct and node.attr in {"os", "_os"}:
-                return ({_OS_ROOT_CHAIN}, set())
-            return ({(*chain, node.attr) for chain in direct}, set())
+            chains = {(*chain, node.attr) for chain in direct}
+            if node.attr in {"os", "_os"}:
+                chains.add(_OS_ROOT_CHAIN)
+            return chains, set()
 
         if isinstance(node, ast.Subscript):
             direct, contained = self._get_os_provenance(node.value)