Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .github/workflows/dotnet-build-and-test.yml
Original file line number Diff line number Diff line change
Expand Up @@ -163,6 +163,7 @@ jobs:
# Change to project directory to ensure local nuget.config is used
pushd consoleapp
dotnet add packcheck.csproj package Microsoft.Agents.AI --prerelease
dotnet add packcheck.csproj package Microsoft.Agents.AI.LocalCodeAct --prerelease
dotnet build -f ${{ matrix.targetFramework }} -c ${{ matrix.configuration }} packcheck.csproj
# Clean up
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -271,6 +271,7 @@ def __init__(
self._allowed_builtins = allowed_builtins if allowed_builtins is not None else ALLOWED_BUILTINS
self._blocked_builtins = blocked_builtins if blocked_builtins is not None else BLOCKED_BUILTINS
self._allowed_os_attrs = allowed_os_attrs if allowed_os_attrs is not None else ALLOWED_OS_ATTRS
self._os_aliases: set[str] = {"os"}

def validate(self, code: str) -> None:
"""Validate code and raise CodeValidationError if it violates policy."""
Expand All @@ -280,6 +281,7 @@ def validate(self, code: str) -> None:
raise CodeValidationError(f"Syntax error in generated code: {exc}") from exc

self._errors = []
self._os_aliases = {"os"}
self.visit(tree)

if self._errors:
Expand All @@ -303,6 +305,10 @@ def visit_Import(self, node: ast.Import) -> None:
self._errors.append(f"Import of '{alias_node.name}' is not allowed (blocked: {module_name})")
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._os_aliases.add(alias_node.asname or "os")
elif alias_node.name.startswith("os.") and alias_node.asname is None:
self._os_aliases.add("os")
self.generic_visit(node)

def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
Expand All @@ -324,6 +330,32 @@ def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
self._errors.append(f"Import from 'os' of '{alias_node.name}' is not allowed")
self.generic_visit(node)

def visit_Assign(self, node: ast.Assign) -> None:
"""Track re-bindings of the ``os`` module."""
for target in node.targets:
self._track_os_alias_targets(target, node.value)
self.generic_visit(node)

def visit_AnnAssign(self, node: ast.AnnAssign) -> None:
"""Track annotated re-bindings of the ``os`` module."""
if (
isinstance(node.value, ast.Name)
and node.value.id in self._os_aliases
and isinstance(node.target, ast.Name)
):
self._os_aliases.add(node.target.id)
self.generic_visit(node)

def _track_os_alias_targets(self, target: ast.AST, value: ast.AST) -> None:
if isinstance(target, ast.Starred):
target = target.value

if isinstance(target, ast.Name) and isinstance(value, ast.Name) and value.id in self._os_aliases:
self._os_aliases.add(target.id)
elif isinstance(target, (ast.Tuple, ast.List)) and isinstance(value, (ast.Tuple, ast.List)):
for target_item, value_item in zip(target.elts, value.elts):
self._track_os_alias_targets(target_item, value_item)

def visit_Call(self, node: ast.Call) -> None:
"""Validate function calls.

Expand Down Expand Up @@ -357,7 +389,7 @@ def visit_Attribute(self, node: ast.Attribute) -> None:
# Enforce the `os` attribute allow-list. Anything outside `ALLOWED_OS_ATTRS`
# (file I/O, process control, mutating helpers, etc.) is rejected so the
# validator matches the documented `os.environ` / `os.path`-only contract.
if isinstance(node.value, ast.Name) and node.value.id == "os" and node.attr not in self._allowed_os_attrs:
if isinstance(node.value, ast.Name) and node.value.id in self._os_aliases and node.attr not in self._allowed_os_attrs:
self._errors.append(f"Access to os.{node.attr} is not allowed")

# Block access to certain dangerous attributes
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,50 @@ await Assert.ThrowsAsync<CodeValidationException>(async () =>
await function.InvokeAsync(args, CancellationToken.None));
}

[Theory]
[InlineData("import os\nos.system('id')")]
[InlineData("import os as x\nx.system('id')")]
[InlineData("import os\n_o = os\n_o.system('id')")]
[InlineData("import os as x\na = x\nb = a\nb.popen('id')")]
[InlineData("import os.path\nos.system('id')")]
[InlineData("import os\na, _ = (os, 1)\na.system('id')")]
[InlineData("import os\n[a, _] = [os, 1]\na.system('id')")]
[InlineData("import os\nx: object = os\nx.system('id')")]
public async Task ExecuteCode_ValidationBlocksDisallowedOsAccessAsync(string code)
{
SkipIfNoPython();

var function = new LocalExecuteCodeFunction(s_python!);

var args = new AIFunctionArguments
{
["code"] = code,
};

var ex = await Assert.ThrowsAsync<CodeValidationException>(async () =>
await function.InvokeAsync(args, CancellationToken.None));
Assert.Contains("os.", ex.Message, StringComparison.Ordinal);
}

[Theory]
[InlineData("import os\nprint(os.environ.get('PATH') is not None)")]
[InlineData("import os as x\nprint(x.path.join('a', 'b'))")]
[InlineData("import os.path as p\nprint(p.join('a', 'b'))")]
public async Task ExecuteCode_AllowsPermittedOsAccessAsync(string code)
{
SkipIfNoPython();

var function = new LocalExecuteCodeFunction(s_python!);

var args = new AIFunctionArguments
{
["code"] = code,
};

var result = await function.InvokeAsync(args, CancellationToken.None);
Assert.NotNull(result);
}

[Fact]
public async Task ExecuteCode_CapturesFilesInWritableMountAsync()
{
Expand Down
Loading