Skip to content
Open
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
52 changes: 44 additions & 8 deletions libs/agno/agno/agents/antigravity/agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -243,8 +243,40 @@ def _build_sources_from_directory(path: Any) -> List[Dict[str, Any]]:
path = Path(path)

sources: List[Dict[str, Any]] = []

def add(file_path: Any, target: str) -> None:
try:
agent_root = path.resolve()
except (OSError, RuntimeError) as e:
log_warning(f"from_agent_directory: cannot resolve {path}: {e}; skipping sources")
return sources

def resolve_source_root(source_root: Any) -> Optional[Path]:
if source_root.is_symlink():
log_warning(f"from_agent_directory: {source_root} is a symlink; skipping")
return None
try:
resolved_root = source_root.resolve()
resolved_root.relative_to(agent_root)
except ValueError:
log_warning(f"from_agent_directory: {source_root} resolves outside {path}; skipping")
return None
except (OSError, RuntimeError) as e:
log_warning(f"from_agent_directory: cannot resolve {source_root}: {e}; skipping")
return None
return resolved_root

def add(file_path: Any, target: str, resolved_root: Any) -> None:
if file_path.is_symlink():
log_warning(f"from_agent_directory: {file_path} is a symlink; skipping")
return
try:
resolved_file = file_path.resolve()
resolved_file.relative_to(resolved_root)
except ValueError:
log_warning(f"from_agent_directory: {file_path} resolves outside its source root; skipping")
return
except (OSError, RuntimeError) as e:
log_warning(f"from_agent_directory: cannot resolve {file_path}: {e}; skipping")
return
try:
size = file_path.stat().st_size
except OSError as e:
Expand All @@ -265,15 +297,19 @@ def add(file_path: Any, target: str) -> None:

workspace = path / "workspace"
if workspace.is_dir():
for f in workspace.rglob("*"):
if f.is_file():
add(f, "/" + f.relative_to(workspace).as_posix())
workspace_root = resolve_source_root(workspace)
if workspace_root is not None:
for f in workspace.rglob("*"):
if f.is_file():
add(f, "/" + f.relative_to(workspace).as_posix(), workspace_root)

skills = path / "skills"
if skills.is_dir():
for f in skills.rglob("*"):
if f.is_file():
add(f, "/.agents/skills/" + f.relative_to(skills).as_posix())
skills_root = resolve_source_root(skills)
if skills_root is not None:
for f in skills.rglob("*"):
if f.is_file():
add(f, "/.agents/skills/" + f.relative_to(skills).as_posix(), skills_root)

return sources

Expand Down
52 changes: 44 additions & 8 deletions libs/agno/agno/tools/antigravity.py
Original file line number Diff line number Diff line change
Expand Up @@ -216,8 +216,40 @@ def _build_sources_from_directory(path: Any) -> List[Dict[str, Any]]:
path = Path(path)

sources: List[Dict[str, Any]] = []

def add(file_path: Any, target: str) -> None:
try:
agent_root = path.resolve()
except (OSError, RuntimeError) as e:
log_warning(f"agent_directory: cannot resolve {path}: {e}; skipping sources")
return sources

def resolve_source_root(source_root: Any) -> Optional[Path]:
if source_root.is_symlink():
log_warning(f"agent_directory: {source_root} is a symlink; skipping")
return None
try:
resolved_root = source_root.resolve()
resolved_root.relative_to(agent_root)
except ValueError:
log_warning(f"agent_directory: {source_root} resolves outside {path}; skipping")
return None
except (OSError, RuntimeError) as e:
log_warning(f"agent_directory: cannot resolve {source_root}: {e}; skipping")
return None
return resolved_root

def add(file_path: Any, target: str, resolved_root: Any) -> None:
if file_path.is_symlink():
log_warning(f"agent_directory: {file_path} is a symlink; skipping")
return
try:
resolved_file = file_path.resolve()
resolved_file.relative_to(resolved_root)
except ValueError:
log_warning(f"agent_directory: {file_path} resolves outside its source root; skipping")
return
except (OSError, RuntimeError) as e:
log_warning(f"agent_directory: cannot resolve {file_path}: {e}; skipping")
return
try:
size = file_path.stat().st_size
except OSError as e:
Expand All @@ -238,15 +270,19 @@ def add(file_path: Any, target: str) -> None:

workspace = path / "workspace"
if workspace.is_dir():
for f in workspace.rglob("*"):
if f.is_file():
add(f, "/" + f.relative_to(workspace).as_posix())
workspace_root = resolve_source_root(workspace)
if workspace_root is not None:
for f in workspace.rglob("*"):
if f.is_file():
add(f, "/" + f.relative_to(workspace).as_posix(), workspace_root)

skills = path / "skills"
if skills.is_dir():
for f in skills.rglob("*"):
if f.is_file():
add(f, "/.agents/skills/" + f.relative_to(skills).as_posix())
skills_root = resolve_source_root(skills)
if skills_root is not None:
for f in skills.rglob("*"):
if f.is_file():
add(f, "/.agents/skills/" + f.relative_to(skills).as_posix(), skills_root)

return sources

Expand Down
36 changes: 36 additions & 0 deletions libs/agno/tests/unit/agents/test_antigravity.py
Original file line number Diff line number Diff line change
Expand Up @@ -367,6 +367,13 @@ def _make_agent_dir(tmp: Path, *, with_agents_md: bool = True, system_instructio
return tmp


def _symlink_or_skip(link: Path, target: Path) -> None:
try:
link.symlink_to(target)
except OSError:
pytest.skip("Symlinks not supported on this system")


def test_from_agent_directory_parses_yaml_and_uses_agents_md_for_instructions():
with tempfile.TemporaryDirectory() as d:
_make_agent_dir(Path(d))
Expand Down Expand Up @@ -402,6 +409,35 @@ def test_from_agent_directory_builds_sources_with_correct_targets():
assert all(s["type"] == "inline" for s in agent.sources)


def test_from_agent_directory_skips_symlinked_workspace_and_skill_files():
with tempfile.TemporaryDirectory() as d:
root = Path(d)
agent_dir = root / "agent"
agent_dir.mkdir()
_make_agent_dir(agent_dir)

outside_workspace = root / "outside_workspace_secret.txt"
outside_skill = root / "outside_skill_secret.txt"
outside_workspace.write_text("AGNO_WORKSPACE_SYMLINK_SECRET")
outside_skill.write_text("AGNO_SKILL_SYMLINK_SECRET")

(agent_dir / "skills" / "leaky").mkdir()
_symlink_or_skip(agent_dir / "workspace" / "linked.txt", outside_workspace)
_symlink_or_skip(agent_dir / "skills" / "leaky" / "SKILL.md", outside_skill)

agent = AntigravityAgent.from_agent_directory(str(agent_dir), api_key="dummy", register=False)

sources = agent.sources or []
targets = {s["target"] for s in sources}
contents = "\n".join(s["content"] for s in sources)
assert "/about.txt" in targets
assert "/.agents/skills/haiku/SKILL.md" in targets
assert "/linked.txt" not in targets
assert "/.agents/skills/leaky/SKILL.md" not in targets
assert "AGNO_WORKSPACE_SYMLINK_SECRET" not in contents
assert "AGNO_SKILL_SYMLINK_SECRET" not in contents


def test_from_agent_directory_requires_id_and_base_agent():
with tempfile.TemporaryDirectory() as d:
p = Path(d)
Expand Down
36 changes: 36 additions & 0 deletions libs/agno/tests/unit/tools/test_antigravity.py
Original file line number Diff line number Diff line change
Expand Up @@ -377,6 +377,13 @@ def _make_toolkit_agent_dir(tmp: Path) -> Path:
return tmp


def _symlink_or_skip(link: Path, target: Path) -> None:
try:
link.symlink_to(target)
except OSError:
pytest.skip("Symlinks not supported on this system")


def test_agent_directory_register_false_parses_without_network():
with tempfile.TemporaryDirectory() as d:
_make_toolkit_agent_dir(Path(d))
Expand All @@ -389,6 +396,35 @@ def test_agent_directory_register_false_parses_without_network():
assert "/.agents/skills/haiku/SKILL.md" in targets


def test_agent_directory_skips_symlinked_workspace_and_skill_files():
with tempfile.TemporaryDirectory() as d:
root = Path(d)
agent_dir = root / "agent"
agent_dir.mkdir()
_make_toolkit_agent_dir(agent_dir)

outside_workspace = root / "outside_workspace_secret.txt"
outside_skill = root / "outside_skill_secret.txt"
outside_workspace.write_text("AGNO_WORKSPACE_SYMLINK_SECRET")
outside_skill.write_text("AGNO_SKILL_SYMLINK_SECRET")

(agent_dir / "skills" / "leaky").mkdir()
_symlink_or_skip(agent_dir / "workspace" / "linked.txt", outside_workspace)
_symlink_or_skip(agent_dir / "skills" / "leaky" / "SKILL.md", outside_skill)

tools = AntigravityTools(api_key="dummy", agent_directory=str(agent_dir), register=False)

sources = tools.default_sources or []
targets = {s["target"] for s in sources}
contents = "\n".join(s["content"] for s in sources)
assert "/about.txt" in targets
assert "/.agents/skills/haiku/SKILL.md" in targets
assert "/linked.txt" not in targets
assert "/.agents/skills/leaky/SKILL.md" not in targets
assert "AGNO_WORKSPACE_SYMLINK_SECRET" not in contents
assert "AGNO_SKILL_SYMLINK_SECRET" not in contents


def test_agent_directory_register_true_posts_to_agents():
captured: List[dict] = []

Expand Down
Loading