@@ -275,7 +275,7 @@ async def resolve_taxonomies(self, context, llm_request) -> list[str]:
275275
276276 plugin = TaxonomyPlugin (policy = SteeringPolicy (), resolver = MockResolver ())
277277
278- # 1. Verify before_model_callback system instruction injection
278+ # Verify before_model_callback system instruction injection
279279 context = mock .MagicMock ()
280280 context .state = {}
281281 llm_request = mock .MagicMock ()
@@ -286,7 +286,7 @@ async def resolve_taxonomies(self, context, llm_request) -> list[str]:
286286 assert context .state ["_active_taxonomies" ] == ["strict" ]
287287 assert llm_request .config .system_instruction == "Original Prompt - MANDATED COMPLIANCE TURN"
288288
289- # 2. Verify skill prioritization/sorting in list_skills
289+ # Verify skill prioritization/sorting in list_skills
290290 skills = [
291291 Skill (frontmatter = Frontmatter (name = "normal" , description = "Desc" ), instructions = "body" ),
292292 Skill (frontmatter = Frontmatter (name = "important" , description = "Desc" ), instructions = "body" ),
@@ -336,14 +336,109 @@ async def test_taxonomy_variable_interpolation():
336336 context = mock .MagicMock ()
337337 context .state = {"_active_taxonomies" : ["urn:adk:domain:finance" ]}
338338
339- # 1. Test shape_description
339+ # Test shape_description
340340 desc = policy .shape_description (skill , context , skill .frontmatter .description )
341341 assert desc == "Read accounts. [PII WARNING]"
342342
343- # 2. Test shape_instructions
343+ # Test shape_instructions
344344 inst = policy .shape_instructions (skill , context , skill .instructions )
345345 assert inst == "Fetch logs.\n Mask SSN"
346346
347- # 3. Test shape_system_instruction
347+ # Test shape_system_instruction
348348 sys_inst = policy .shape_system_instruction (context , ["urn:adk:domain:finance" ], "Start. {taxonomy:warning}" )
349349 assert sys_inst == "Start. [PII WARNING]"
350+
351+
352+ @pytest .mark .asyncio
353+ async def test_taxonomy_plugin_path_validation ():
354+ """Tests that absolute paths and path traversals are blocked on all platforms."""
355+ plugin = TaxonomyPlugin ()
356+ mock_tool = mock .MagicMock ()
357+ mock_tool .name = "load_skill"
358+ context = mock .MagicMock ()
359+ context .state = {}
360+
361+ # Test cases for blocked paths (absolute or traversal)
362+ blocked_cases = [
363+ "/etc/passwd" ,
364+ "C:\\ Windows\\ System32" ,
365+ "\\ \\ unc\\ share\\ file" ,
366+ "../../traversal" ,
367+ "subdir\\ ..\\ parent" ,
368+ ]
369+
370+ for file_path in blocked_cases :
371+ result = await plugin .before_tool_callback (
372+ tool = mock_tool ,
373+ tool_args = {"skill_name" : "test-skill" , "file_path" : file_path },
374+ tool_context = context ,
375+ )
376+ assert isinstance (result , dict )
377+ assert result .get ("error_code" ) == "INVALID_ARGUMENTS"
378+ assert "blocked" in result .get ("error" , "" ).lower ()
379+
380+
381+ @pytest .mark .asyncio
382+ async def test_taxonomy_variable_interpolation_warning (caplog ):
383+ """Tests that unresolved taxonomy variables log a warning and fallback to empty string."""
384+ registry = TaxonomyRegistry (terms = {})
385+ policy = DefaultSkillPolicy (registry )
386+ context = mock .MagicMock ()
387+ context .state = {"_active_taxonomies" : ["urn:adk:domain:test" ]}
388+
389+ with caplog .at_level ("WARNING" ):
390+ result = policy .shape_system_instruction (
391+ context , ["urn:adk:domain:test" ], "Prompt: {taxonomy:missing_variable}"
392+ )
393+
394+ assert result == "Prompt: "
395+ assert len (caplog .records ) == 1
396+ assert "missing_variable" in caplog .records [0 ].message
397+
398+
399+ def test_taxonomy_custom_shape_skill ():
400+ """Tests default sanitization and custom policy shape_skill overriding."""
401+
402+ from pydantic import Field
403+
404+ class ExtendedFrontmatter (Frontmatter ):
405+ billing_entitlement : str = Field ("premium" , alias = "billingEntitlement" )
406+ custom_flag : bool = True
407+
408+ original_skill = Skill (
409+ frontmatter = ExtendedFrontmatter (
410+ name = "custom-skill" ,
411+ description = "My skill" ,
412+ billingEntitlement = "enterprise" ,
413+ ),
414+ instructions = "Execute tasks"
415+ )
416+
417+ policy = DefaultSkillPolicy ()
418+ context = mock .MagicMock ()
419+
420+ # It should drop the custom pydantic field billing_entitlement because it's not captured by standard Frontmatter
421+ default_shaped = policy .shape_skill (original_skill , context , "Shaped My skill" )
422+ assert default_shaped .frontmatter .name == "custom-skill"
423+ assert default_shaped .frontmatter .description == "Shaped My skill"
424+ # Standard Frontmatter doesn't have custom_flag or billing_entitlement as defined properties
425+ assert not hasattr (default_shaped .frontmatter , "custom_flag" )
426+ assert not hasattr (default_shaped .frontmatter , "billing_entitlement" )
427+
428+ # Verify Custom Policy Behavior using model_copy
429+ class CustomCopyPolicy (DefaultSkillPolicy ):
430+ def shape_skill (self , skill , context , shaped_description ):
431+ new_fm = skill .frontmatter .model_copy (update = {"description" : shaped_description })
432+ return skill .model_copy (update = {"frontmatter" : new_fm })
433+
434+ custom_policy = CustomCopyPolicy ()
435+ custom_shaped = custom_policy .shape_skill (original_skill , context , "Shaped My skill" )
436+
437+ # Ensure all custom attributes, types, and values are fully preserved!
438+ assert custom_shaped .frontmatter .name == "custom-skill"
439+ assert custom_shaped .frontmatter .description == "Shaped My skill"
440+ assert isinstance (custom_shaped .frontmatter , ExtendedFrontmatter )
441+ assert custom_shaped .frontmatter .billing_entitlement == "enterprise"
442+ assert custom_shaped .frontmatter .custom_flag is True
443+
444+
0 commit comments