Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/discovery_endpoints/agent_skills_archive.py: 27%

1"""Repack a stored skill upload into the archive shape Agent Skills clients install from. 

2 

3Uploads follow the Anthropic Skills API layout, where every file sits under a single 

4top-level folder. Discovery clients read ``SKILL.md`` from the archive root, so that 

5folder is stripped and the zip is rebuilt with fixed entry timestamps, which keeps the 

6SHA-256 digest published in the index reproducible for identical uploads. 

7""" 

8 

9import hashlib 

10import io 

11import re 

12import zipfile 

13from collections.abc import Mapping 

14from dataclasses import dataclass 

15from types import MappingProxyType 

16from typing import Final 

17 

18import yaml 

19 

20MAX_ARCHIVE_UNPACKED_BYTES: Final = 50 * 1024 * 1024 

21MAX_ARCHIVE_ENTRIES: Final = 1000 

22SKILL_MANIFEST_FILENAME: Final = "SKILL.md" 

23 

24_ZIP_ENTRY_TIMESTAMP: Final = (1980, 1, 1, 0, 0, 0) 

25_ZIP_ENTRY_PERMISSIONS: Final = 0o644 << 16 

26_FRONTMATTER_PATTERN: Final = re.compile(r"^---\s*\n(.*?)\n---\s*(?:\n|$)", re.DOTALL) 

27_WINDOWS_DRIVE_PATTERN: Final = re.compile(r"^[A-Za-z]:") 

28_EMPTY_FRONTMATTER: Final[Mapping[str, object]] = MappingProxyType({}) 

29 

30 

31@dataclass(frozen=True, slots=True) 

32class SkillArchive: 

33 content: bytes 

34 digest: str 

35 declared_name: str | None 

36 declared_description: str | None 

37 

38 

39def build_skill_archive(stored_content: bytes) -> SkillArchive | None: 

40 """Return the installable archive for an upload, or None when it holds no root SKILL.md.""" 

41 try: 

42 with zipfile.ZipFile(io.BytesIO(stored_content)) as uploaded: 

43 members: Final = _flattened_members(uploaded) 

44 except (zipfile.BadZipFile, OSError, RuntimeError): 

45 return None 

46 

47 if members is None: 

48 return None 

49 

50 frontmatter: Final = _manifest_frontmatter(next(data for name, data in members if name == SKILL_MANIFEST_FILENAME)) 

51 content: Final = _repack(members) 

52 return SkillArchive( 

53 content=content, 

54 digest=f"sha256:{hashlib.sha256(content).hexdigest()}", 

55 declared_name=_frontmatter_text(frontmatter, "name"), 

56 declared_description=_frontmatter_text(frontmatter, "description"), 

57 ) 

58 

59 

60def _flattened_members(uploaded: zipfile.ZipFile) -> tuple[tuple[str, bytes], ...] | None: 

61 infos: Final = tuple(info for info in uploaded.infolist() if not info.is_dir()) 

62 if not infos or len(infos) > MAX_ARCHIVE_ENTRIES: 

63 return None 

64 if sum(info.file_size for info in infos) > MAX_ARCHIVE_UNPACKED_BYTES: 

65 return None 

66 

67 normalized: Final = tuple((info, _normalized_path(info.filename)) for info in infos) 

68 if any(path is None for _, path in normalized): 

69 return None 

70 

71 prefix: Final = _common_root_prefix(tuple(path for _, path in normalized if path is not None)) 

72 flattened: Final = tuple((info, path[len(prefix) :]) for info, path in normalized if path is not None) 

73 names: Final = frozenset(name for _, name in flattened) 

74 if SKILL_MANIFEST_FILENAME not in names or len(names) != len(flattened): 

75 return None 

76 

77 return tuple((name, uploaded.read(info)) for info, name in sorted(flattened, key=lambda member: member[1])) 

78 

79 

80def _common_root_prefix(paths: tuple[str, ...]) -> str: 

81 roots: Final = frozenset(path.split("/", 1)[0] for path in paths) 

82 if len(roots) != 1 or not all("/" in path for path in paths): 

83 return "" 

84 return f"{next(iter(roots))}/" 

85 

86 

87def _normalized_path(raw_path: str) -> str | None: 

88 if not raw_path or "\0" in raw_path or "\\" in raw_path: 

89 return None 

90 if raw_path.startswith("/") or _WINDOWS_DRIVE_PATTERN.match(raw_path): 

91 return None 

92 parts: Final = tuple(part for part in raw_path.split("/") if part) 

93 if not parts or any(part in (".", "..") for part in parts): 

94 return None 

95 return "/".join(parts) 

96 

97 

98def _manifest_frontmatter(manifest: bytes) -> Mapping[str, object]: 

99 match: Final = _FRONTMATTER_PATTERN.match(manifest.decode("utf-8", errors="replace")) 

100 if match is None: 

101 return _EMPTY_FRONTMATTER 

102 try: 

103 parsed: Final = yaml.safe_load(match.group(1)) 

104 except yaml.YAMLError: 

105 return _EMPTY_FRONTMATTER 

106 if not isinstance(parsed, dict): 

107 return _EMPTY_FRONTMATTER 

108 return parsed 

109 

110 

111def _frontmatter_text(frontmatter: Mapping[str, object], key: str) -> str | None: 

112 value: Final = frontmatter.get(key) 

113 if not isinstance(value, str): 

114 return None 

115 return value.strip() or None 

116 

117 

118def _zip_entry(name: str) -> zipfile.ZipInfo: 

119 entry: Final = zipfile.ZipInfo(filename=name, date_time=_ZIP_ENTRY_TIMESTAMP) 

120 entry.compress_type = zipfile.ZIP_DEFLATED 

121 entry.external_attr = _ZIP_ENTRY_PERMISSIONS 

122 return entry 

123 

124 

125def _repack(members: tuple[tuple[str, bytes], ...]) -> bytes: 

126 buffer: Final = io.BytesIO() 

127 with zipfile.ZipFile(buffer, mode="w", compression=zipfile.ZIP_DEFLATED) as repacked: 

128 for name, data in members: 

129 repacked.writestr(_zip_entry(name), data) 

130 return buffer.getvalue()