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
2 changes: 1 addition & 1 deletion experiments/run_experiment.py
Original file line number Diff line number Diff line change
Expand Up @@ -130,7 +130,7 @@ def _main(cfg: DictConfig) -> float:
desc_path.write_text(description)
env_description_path = str(desc_path)

primitives = build_primitives(env, cfg.primitives)
primitives = build_primitives(env, cfg.primitives, blackbox=blackbox)

# Write env config for MCP server (if mcp_tools are configured).
mcp_tools = tuple(cfg.get("mcp_tools", []))
Expand Down
16 changes: 13 additions & 3 deletions experiments/tracker/constraints.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,15 +15,25 @@ class ExperimentConfig:
replicate_seeds: tuple[int, ...]


# Primitive levels that grant a primitive bound to the live environment (see
# ENV_DEPENDENT_PRIMITIVES: bilevel grants bilevel_models, low_level grants
# check_action_collision). Such a primitive closes over the env at eval time, so a
# black-box program granted one could read the environment out of its closure --
# build_primitives refuses the combination, and campaigns must not schedule it.
_ENV_BOUND_PRIMITIVE_LEVELS = frozenset({"bilevel", "low_level"})


def is_valid_experiment(config: ExperimentConfig) -> tuple[bool, str | None]:
"""Return whether a Hydra-canonicalized condition is runnable."""
blackbox = config.values.get("approach.blackbox") is True
strict = config.values.get("approach.blackbox_strict") is True
bilevel = config.values.get("primitive_level") == "bilevel"
level = config.values.get("primitive_level")
if strict and not blackbox:
return False, "strict blackbox runtime requires blackbox access"
if strict and config.values.get("primitive_level") != "none":
if strict and level != "none":
return False, "strict blackbox runtime requires primitive_level=none"
if blackbox and bilevel:
if blackbox and level == "bilevel":
return False, "bilevel_models unavailable under blackbox"
if blackbox and level in _ENV_BOUND_PRIMITIVE_LEVELS:
return False, "env-bound primitives unavailable under blackbox"
return True, None
3 changes: 3 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,9 @@ target-version = ["py311"]
[tool.isort]
py_version = 311
profile = "black"
# isort 7 classifies `_thread` as third-party and isort 9 as stdlib; CI floats to the
# latest, so pin the classification rather than let the two rewrite each other.
extra_standard_library = ["_thread"]
multi_line_output = 2
skip_glob = ["venv/*", ".venv/*"]
split_on_trailing_comma = true
Expand Down
30 changes: 28 additions & 2 deletions src/robocode/primitives/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
from robocode.primitives.check_action_collision import check_action_collision
from robocode.primitives.motion_planning import BiRRT
from robocode.utils.bilevel import build_sesame_models
from robocode.utils.scored_env_guard import readonly_view

__all__ = [
"ENV_DEPENDENT_PRIMITIVES",
Expand Down Expand Up @@ -70,8 +71,33 @@
)


def build_primitives(env: Any, names: list[str] | tuple[str, ...]) -> dict[str, Any]:
"""Build a primitives dict containing only the requested *names*."""
def build_primitives(
env: Any, names: list[str] | tuple[str, ...], *, blackbox: bool = False
) -> dict[str, Any]:
"""Build a primitives dict containing only the requested *names*.

Env-bound primitives close over a read-only view of *env*: a generated approach that
receives one could otherwise reach the environment it is being scored in through the
primitive's closure and move it into a solved state.

The view stops writes but deliberately passes reads through, which is right under
whitebox (the agent has the env source anyway) and wrong under *blackbox*, where the
whole point is that the program never sees the env. Reads cannot be closed off
in-process -- a closure's contents are reachable whatever the binding looks like --
so the combination is refused here rather than served unsafely. Lifting this means
proxying env-bound primitives through the env server at eval time, the way
``blackbox_primitive_manifest`` already does for the sandbox.
"""
if blackbox:
env_bound = [n for n in names if n in ENV_DEPENDENT_PRIMITIVES]
if env_bound:
raise ValueError(
f"Primitives {', '.join(sorted(env_bound))} are bound to the live "
"environment, so a black-box approach granted one could read the "
"environment it is being scored in out of the primitive's closure. "
"Run these whitebox, or use primitive_level=none."
)
env = readonly_view(env)
for name in names:
if name in _DEPRECATED_PRIMITIVES:
logger.warning(
Expand Down
44 changes: 16 additions & 28 deletions src/robocode/utils/episode.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@
import logging
import multiprocessing as mp
import os
import re
import signal
import subprocess
import sys
Expand All @@ -24,7 +23,11 @@
from numpy.typing import NDArray

from robocode.approaches.base_approach import BaseApproach
from robocode.utils.strict_blackbox import check_strict_imports, module_locations
from robocode.utils.strict_blackbox import (
check_strict_imports,
module_locations,
reachable_sibling_files,
)

logger = logging.getLogger(__name__)

Expand All @@ -41,7 +44,7 @@
)


def _reject_planner_references(source: str, primitives: dict[str, Any]) -> None:
def _reject_planner_references(path: Path, primitives: dict[str, Any]) -> None:
"""Reject a generated approach that invokes the bilevel planner (anti-cheat).

Only enforced when the ``bilevel_models`` primitive is in play; other
Expand All @@ -52,19 +55,16 @@ def _reject_planner_references(source: str, primitives: dict[str, Any]) -> None:
# A cooperative guardrail plus a clear error message, not an adversarial
# sandbox: substring matching suffices because aliased imports still carry the
# module path, and only deliberate importlib obfuscation would evade it.
hits = [ref for ref in _FORBIDDEN_PLANNER_REFS if ref in source]
if hits:
raise ValueError(
f"Generated approach references the bilevel planner ({', '.join(hits)}); "
"with the bilevel_models primitive you must compose the models yourself, "
"not run SeSamE search."
)

for source_path in reachable_sibling_files(path):
source = source_path.read_text(encoding="utf-8")
hits = [ref for ref in _FORBIDDEN_PLANNER_REFS if ref in source]
if hits:
raise ValueError(
f"Generated approach references the bilevel planner "
f"({', '.join(hits)}) in {source_path.name}; with the bilevel_models "
"primitive you must compose the models yourself, not run SeSamE search."
)

# A frozen GeneratedApproach is scored through reset()/get_action() only; referencing
# set_state or sample_next_state mutates the scored env and can fake a solve. Match
# ".name" with an identifier boundary so get_state, reset_state, and set_stateful pass.
_FORBIDDEN_STATE_MUTATIONS = ("set_state", "sample_next_state")

# The sibling modules a generated policy imported at load time stay cached while
# it is active (its methods may import them again) and are removed before the next
Expand All @@ -73,17 +73,6 @@ def _reject_planner_references(source: str, primitives: dict[str, Any]) -> None:
_LOADED_GENERATED_SIBLINGS: dict[str, ModuleType] = {}


def _reject_state_mutation(source: str) -> None:
"""Reject a generated approach that mutates the scored env (anti-cheat)."""
hits = [n for n in _FORBIDDEN_STATE_MUTATIONS if re.search(rf"\.{n}\b", source)]
if hits:
raise ValueError(
f"Generated approach references {', '.join(hits)}; approach.py is scored "
"through reset()/get_action() only and must reach the goal via the actions "
"it returns, not by mutating the environment's state."
)


def _evict_loaded_generated_siblings() -> None:
"""Remove sibling modules retained by the previous generated policy load."""
for name, module in _LOADED_GENERATED_SIBLINGS.items():
Expand Down Expand Up @@ -128,9 +117,8 @@ def load_generated_approach(
sys.path.insert(0, sandbox_dir)
modules_before = set(sys.modules)
try:
_reject_planner_references(path, primitives)
source = path.read_text()
_reject_planner_references(source, primitives)
_reject_state_mutation(source)
# Set __file__ so the exec'd code can use it (e.g. to locate
# sibling modules via os.path.dirname(__file__)). exec() does
# not set this automatically unlike a normal module import.
Expand Down
92 changes: 92 additions & 0 deletions src/robocode/utils/scored_env_guard.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,92 @@
"""Read-only view of the scored environment, for handing to generated code.

A frozen ``GeneratedApproach`` is scored through ``reset()``/``get_action()``: it must
reach the goal via the actions it returns, not by moving the environment into a
solved state. The approach is not given the environment directly, but some primitives
are bound to it (``check_action_collision`` is a ``partial`` over the live env), so a
program that receives such a primitive can reach the scored environment through its
closure.

:func:`readonly_view` wraps the environment so that the mutating entry points raise
instead. Everything else -- reading state, geometry handles, collision queries --
passes through untouched.

This replaces a source scan that looked for ``.set_state`` in ``approach.py``. That
check was wrong in both directions: it missed the same call in a sibling module the
approach imports, and it rejected programs that build a *private* environment of
their own to plan in, which is the ordinary way to write a TAMP policy and cannot
affect scoring at all. Guarding the object the approach can actually reach is exact:
a private clone is unaffected, and no amount of aliasing gets around it.
"""

from __future__ import annotations

from typing import Any, NoReturn

# Mutators that could move the scored environment into a solved state, or desync it
# from the episode the runner is scoring.
FORBIDDEN = frozenset({"set_state", "sample_next_state", "reset", "step", "close"})


class ScoredEnvMutationError(RuntimeError):
"""Raised when generated code tries to mutate the environment being scored."""


class _ReadOnlyEnv:
"""Attribute proxy that forwards everything except the mutating entry points."""

__slots__ = ("_env",)

def __init__(self, env: Any) -> None:
object.__setattr__(self, "_env", env)

def __getattribute__(self, name: str) -> Any:
# Primitives dispatch on isinstance(env, ...) to pick an implementation, so
# the proxy reports the wrapped environment's class; a proxy advertising its
# own type would fall through to "unsupported environment". Only the type is
# borrowed -- every other attribute still routes through the guard below.
if name == "__class__":
return object.__getattribute__(self, "_env").__class__
return object.__getattribute__(self, name)

@property
def unwrapped_scored_env(self) -> Any:
"""The wrapped environment, for host-side code that legitimately mutates it."""
return object.__getattribute__(self, "_env")

def __getattr__(self, name: str) -> Any:
if name in FORBIDDEN:
return _forbid(name)
return getattr(object.__getattribute__(self, "_env"), name)

def __setattr__(self, name: str, value: Any) -> NoReturn:
raise ScoredEnvMutationError(
f"Cannot set {name!r} on the environment being scored. Build your own "
"environment to plan in; this one is scored through the actions your "
"approach returns."
)

def __repr__(self) -> str:
return f"<read-only {object.__getattribute__(self, '_env')!r}>"


def _forbid(name: str) -> Any:
def _raise(*_args: Any, **_kwargs: Any) -> NoReturn:
raise ScoredEnvMutationError(
f"{name}() is not available on the environment being scored: an approach "
"must reach the goal through the actions it returns, not by moving the "
"environment. To plan, construct your own environment and mutate that."
)

return _raise


def readonly_view(env: Any) -> Any:
"""Wrap *env* so generated code cannot mutate it.

Idempotent.
"""
# `type(...) is` rather than isinstance: the proxy reports the wrapped class.
if type(env) is _ReadOnlyEnv: # pylint: disable=unidiomatic-typecheck
return env
return _ReadOnlyEnv(env)
37 changes: 37 additions & 0 deletions src/robocode/utils/strict_blackbox.py
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,43 @@ def _relative_sibling_files(
return files


def reachable_sibling_files(entry: Path) -> list[Path]:
"""Every sandbox file reachable from *entry* through sibling imports, plus *entry*.

``load_generated_approach`` puts the sandbox directory on ``sys.path`` so that
``approach.py`` can import siblings the agent wrote, which means a check that reads
only ``approach.py`` is bypassed by moving the code one file over. Callers that
scan program source (the bilevel planner check) walk this instead of a single file.

Shares the import-resolution helpers with :func:`check_strict_imports` but applies
no allowlist: it reports what the program is made of, not whether it is permitted.
Unparseable files are yielded without contributing further edges.
"""
root = entry.resolve().parent
pending = [entry.resolve()]
seen: list[Path] = []
visited: set[Path] = set()
while pending:
path = pending.pop()
if path in visited:
continue
visited.add(path)
seen.append(path)
try:
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
except (SyntaxError, OSError):
continue
for node in ast.walk(tree):
if not isinstance(node, (ast.Import, ast.ImportFrom)):
continue
if isinstance(node, ast.ImportFrom) and node.level:
pending.extend(_relative_sibling_files(root, path, node) or [])
continue
for name in _import_names(node):
pending.extend(_sibling_files(root, name.split(".", 1)[0]))
return seen


def _import_names(node: ast.Import | ast.ImportFrom) -> list[str]:
if isinstance(node, ast.Import):
return [alias.name for alias in node.names]
Expand Down
13 changes: 9 additions & 4 deletions tests/experiments/tracker/test_generate.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,13 +41,18 @@ def _config(**updates: Any) -> Any:


def test_sample_campaign_expansion_and_constraint(tmp_path: Path) -> None:
"""The sample campaign yields five rows and excludes one invalid matrix cell."""
"""The sample campaign yields four rows and excludes two invalid matrix cells."""
rows, excluded = generate.generate_rows(
[_sample_campaign(tmp_path)], eval_seed=_TEST_EVAL_SEED
)
assert len(rows) == 5
assert len(excluded) == 1
assert excluded[0][1] == "bilevel_models unavailable under blackbox"
assert len(rows) == 4
# Blackbox excludes both env-bound primitive levels: bilevel has no host proxy at
# all, and low_level's check_action_collision would let the program read the env
# out of the primitive's closure at eval time.
assert sorted(reason for _, reason, _ in excluded) == [
"bilevel_models unavailable under blackbox",
"env-bound primitives unavailable under blackbox",
]
assert all(row["Replicate Seeds"] == "[42, 24]" for row in rows)
assert all(row["Evaluation Seed"] == str(_TEST_EVAL_SEED) for row in rows)
assert all("replicate_seed=42,24" in row["Command"] for row in rows)
Expand Down
20 changes: 20 additions & 0 deletions tests/primitives/test_bilevel_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -138,3 +138,23 @@ def test_variable_count_symbolic_layer() -> None:
"PlaceOnTable",
"PlaceOnTarget",
}


def test_blackbox_refuses_env_bound_primitives() -> None:
"""A black-box program must not be able to read the env it is scored in.

Env-bound primitives close over the live env, and the read-only view passes reads
through, so granting one under blackbox would hand the program the environment the
mode exists to withhold. Refused rather than served unsafely.
"""
env = _obstruction_env()
with pytest.raises(ValueError, match="black-box"):
build_primitives(env, ["check_action_collision"], blackbox=True)
with pytest.raises(ValueError, match="black-box"):
build_primitives(env, ["bilevel_models"], blackbox=True)


def test_blackbox_allows_generic_primitives() -> None:
"""Generic primitives carry no env, so blackbox is free to grant them."""
env = _obstruction_env()
assert "BiRRT" in build_primitives(env, ["BiRRT"], blackbox=True)
Loading
Loading