Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
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 Down Expand Up @@ -303,6 +304,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 +329,24 @@ 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."""
if isinstance(node.value, ast.Name) and node.value.id in self._os_aliases:
for target in node.targets:
if isinstance(target, ast.Name):
self._os_aliases.add(target.id)
Comment thread
eavanvalkenburg marked this conversation as resolved.
Outdated
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 visit_Call(self, node: ast.Call) -> None:
"""Validate function calls.

Expand Down Expand Up @@ -357,7 +380,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,47 @@ 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')")]
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