|
17 | 17 | from __future__ import annotations |
18 | 18 |
|
19 | 19 | import logging |
20 | | -from typing import Any |
21 | | -from typing import Optional |
| 20 | +from pathlib import PurePosixPath, PureWindowsPath |
| 21 | +from typing import Any, Optional |
22 | 22 |
|
23 | 23 | from google.adk.plugins.base_plugin import BasePlugin |
24 | 24 | from google.adk.agents.callback_context import CallbackContext |
|
28 | 28 | from google.adk.tools.base_tool import BaseTool |
29 | 29 | from google.adk.tools.tool_context import ToolContext |
30 | 30 |
|
| 31 | +from .policy import DefaultSkillPolicy |
31 | 32 | from .policy import SkillPolicy |
32 | 33 | from .policy import TaxonomyResolver |
33 | 34 | from .taxonomy_config import TaxonomyRegistry |
@@ -106,6 +107,11 @@ async def before_tool_callback( |
106 | 107 | if tool.name not in _SKILL_GATE_TOOLS: |
107 | 108 | return None |
108 | 109 |
|
| 110 | + # Rule 5 Assertions |
| 111 | + assert tool is not None, "Intercepted tool cannot be None" |
| 112 | + assert isinstance(tool_args, dict), "tool_args must be a dictionary" |
| 113 | + assert tool_context is not None, "tool_context cannot be None" |
| 114 | + |
109 | 115 | active_taxonomies = ( |
110 | 116 | tool_context.state.get(_ACTIVE_TAXONOMIES_STATE_KEY) or [] |
111 | 117 | ) |
@@ -133,7 +139,18 @@ async def before_tool_callback( |
133 | 139 |
|
134 | 140 | file_path = tool_args.get("file_path") |
135 | 141 | if file_path: |
136 | | - if ".." in file_path or file_path.startswith(("/", "\\")): |
| 142 | + posix_p = PurePosixPath(file_path) |
| 143 | + win_p = PureWindowsPath(file_path) |
| 144 | + |
| 145 | + # Block absolute paths or presence of a drive letter |
| 146 | + if posix_p.is_absolute() or win_p.is_absolute() or win_p.drive: |
| 147 | + return { |
| 148 | + "error": f"Absolute path blocked: {file_path}", |
| 149 | + "error_code": "INVALID_ARGUMENTS", |
| 150 | + } |
| 151 | + |
| 152 | + # Block traversal segments |
| 153 | + if ".." in posix_p.parts or ".." in win_p.parts: |
137 | 154 | return { |
138 | 155 | "error": f"Path traversal attempt blocked: {file_path}", |
139 | 156 | "error_code": "INVALID_ARGUMENTS", |
@@ -187,21 +204,11 @@ def _filter_list_skills( |
187 | 204 | allowed_skills, tool_context, active_taxonomies |
188 | 205 | ) |
189 | 206 |
|
190 | | - from google.adk.skills.models import Skill, Frontmatter |
191 | | - |
192 | 207 | shaped_skills = [] |
193 | 208 | for skill in prioritized_skills: |
194 | 209 | original_desc = skill.frontmatter.description or "" |
195 | 210 | shaped_desc = self.policy.shape_description(skill, tool_context, original_desc) |
196 | | - extra = getattr(skill.frontmatter, "model_extra", None) or {} |
197 | | - new_skill = Skill( |
198 | | - frontmatter=Frontmatter( |
199 | | - name=skill.frontmatter.name, |
200 | | - description=shaped_desc, |
201 | | - **extra |
202 | | - ), |
203 | | - instructions=skill.instructions |
204 | | - ) |
| 211 | + new_skill = self.policy.shape_skill(skill, tool_context, shaped_desc) |
205 | 212 | shaped_skills.append(new_skill) |
206 | 213 |
|
207 | 214 | logger.debug( |
|
0 commit comments