Coverage for src / osiris_cli / tool_policy.py: 0%

1import re 

2from typing import Dict, List, Optional, Tuple 

3 

4from .session import SessionManager, session 

5from .tools import tools 

6from .tools_registry import tools_registry 

7 

8READ_ONLY_TOOL_SET = { 

9 "read_file", 

10 "list_directory", 

11 "get_file_tree", 

12 "search_codebase", 

13} 

14 

15def _build_tool_name_map() -> Dict[str, str]: 

16 names: Dict[str, str] = {} 

17 for tool in tools_registry.list_tools(): 

18 key = tool.name.lower() 

19 names[key] = tool.name 

20 for tool_name in tools.tools.keys(): 

21 key = tool_name.lower() 

22 names.setdefault(key, tool_name) 

23 return names 

24 

25 

26def _parse_tool_tokens(fragment: str, name_map: Dict[str, str]) -> List[str]: 

27 tokens = re.findall(r"[a-zA-Z0-9_]+", fragment) 

28 parsed: List[str] = [] 

29 for token in tokens: 

30 canonical = name_map.get(token.lower()) 

31 if canonical and canonical not in parsed: 

32 parsed.append(canonical) 

33 return parsed 

34 

35 

36def extract_tool_constraints(text: str) -> Tuple[List[str], List[str], bool]: 

37 allowed: List[str] = [] 

38 blocked: List[str] = [] 

39 name_map = _build_tool_name_map() 

40 header_pattern = re.compile( 

41 r"(allowed tools|blocked tools|forbidden tools|allowed|blocked|forbidden)\s*:", 

42 re.IGNORECASE 

43 ) 

44 matches = list(header_pattern.finditer(text)) 

45 

46 for idx, match in enumerate(matches): 

47 header = match.group(1).lower() 

48 section = "allowed" if "allow" in header else "blocked" 

49 start = match.end() 

50 end = matches[idx + 1].start() if idx + 1 < len(matches) else len(text) 

51 segment = text[start:end] 

52 tokens = _parse_tool_tokens(segment, name_map) 

53 if section == "allowed": 

54 allowed.extend(tokens) 

55 else: 

56 blocked.extend(tokens) 

57 

58 read_only_flag = bool(re.search(r"read[-\s]?only", text, re.IGNORECASE)) 

59 return allowed, blocked, read_only_flag 

60 

61 

62def apply_tool_constraints(text: str, target_session: Optional[SessionManager] = None) -> bool: 

63 session_obj = target_session or session 

64 allowed, blocked, read_only_flag = extract_tool_constraints(text) 

65 if read_only_flag and not allowed: 

66 name_map = _build_tool_name_map() 

67 allowed = [name_map[name] for name in READ_ONLY_TOOL_SET if name in name_map] 

68 if not allowed and not blocked: 

69 return False 

70 policy_updated = session_obj.apply_tool_policy( 

71 allowed=allowed if allowed else None, 

72 forbidden=blocked if blocked else None 

73 ) 

74 if policy_updated: 

75 session_obj.add_activity("Applied tool policy from request.") 

76 return policy_updated