Coverage for plugin/scripts/_rhiza_bundles.py: 100%

1"""Resolve profiles and bundles to the file paths a sync should copy. 

2 

3The largest single job in the sync and the one with no I/O at all: `profiles` expand to 

4bundle names, bundle names expand to `(source, dest)` file entries, and the result is an 

5ordered, de-duplicated path list plus a remap table. 

6 

7It also applies the path-safety check. `template-bundles.yml` comes from the template 

8repo, so a `dest` is untrusted input that gets joined onto the target directory — an 

9absolute path, a drive letter or a `..` component would write outside the project. The 

10rule is `_rhiza_common.escapes_root`, shared with `stage_synced.py`, which judges lock 

11entries by exactly the same standard; what this module owns is aborting the sync over one. 

12""" 

13 

14from __future__ import annotations 

15 

16import sys 

17from dataclasses import dataclass 

18from pathlib import Path 

19from typing import Any 

20 

21sys.path.insert(0, str(Path(__file__).resolve().parent)) 

22from _rhiza_common import SyncError, escapes_root # noqa: E402 

23from _rhiza_template import Template # noqa: E402 

24from _rhiza_yaml import as_list # noqa: E402 

25 

26 

27def _ensure_safe_bundle_path(value: str) -> None: 

28 r"""Reject a bundle path that could escape the project directory. 

29 

30 ``template-bundles.yml`` is untrusted (fetched from the template repo) and a 

31 remapped ``dest`` is joined onto the target directory, so an absolute path, a 

32 Windows drive letter, or a ``..`` component could write outside the project. 

33 

34 The three rejected shapes, against the ordinary relative paths that must keep 

35 working — and note the fourth line, where a Windows separator is normalised *before* 

36 the check, so a backslash cannot smuggle a traversal past it: 

37 

38 >>> for value in ("Makefile", ".github/workflows/ci.yml", "/etc/passwd", 

39 ... "..\\secrets.env", "C:/Windows/system32"): 

40 ... try: 

41 ... _ensure_safe_bundle_path(value) 

42 ... print(f"accepted {value}") 

43 ... except SyncError: 

44 ... print(f"rejected {value}") 

45 accepted Makefile 

46 accepted .github/workflows/ci.yml 

47 rejected /etc/passwd 

48 rejected ..\secrets.env 

49 rejected C:/Windows/system32 

50 

51 The rule itself lives in `_rhiza_common.escapes_root`, because a lock file's ``files`` 

52 entry is joined onto the target the same way and must be judged identically — see 

53 `stage_synced.py`. What stays here is the *consequence*: a bad bundle path aborts the 

54 sync, while a bad lock entry is reported by a script that returns an exit code. 

55 

56 Raises: 

57 SyncError: If *value* is absolute, uses a drive letter, or traverses up. 

58 """ 

59 if escapes_root(value): 

60 raise SyncError( 

61 f"Unsafe bundle path {value!r}: paths must be relative to the project root " 

62 "(no absolute paths, drive letters, or '..' traversal)." 

63 ) 

64 

65 

66def _bundle_file_entries(raw_files: Any) -> list[tuple[str, str]]: 

67 """Coerce a bundle's ``files`` field into validated ``(source, dest)`` pairs.""" 

68 entries: list[tuple[str, str]] = [] 

69 for entry in as_list(raw_files) if isinstance(raw_files, str) else (raw_files or []): 

70 if isinstance(entry, str): 

71 source = dest = entry 

72 elif isinstance(entry, dict) and "source" in entry: 

73 source = str(entry["source"]) 

74 dest = str(entry.get("dest", source)) 

75 else: 

76 raise SyncError( 

77 f"Bundle file entry must be a string or a {{source, dest}} map, got: {entry!r}" 

78 ) 

79 _ensure_safe_bundle_path(source) 

80 _ensure_safe_bundle_path(dest) 

81 entries.append((source, dest)) 

82 return entries 

83 

84 

85@dataclass(frozen=True) 

86class Bundles: 

87 """The bundle/profile definitions from `template-bundles.yml` that sync needs.""" 

88 

89 requires: dict[str, list[str]] 

90 files: dict[str, list[tuple[str, str]]] 

91 profiles: dict[str, list[str]] 

92 

93 @classmethod 

94 def from_config(cls, config: dict[str, Any]) -> Bundles: 

95 """Parse a `template-bundles.yml` dict into requires/files/profiles maps.""" 

96 raw_bundles = config.get("bundles") or {} 

97 raw_profiles = config.get("profiles") or {} 

98 requires: dict[str, list[str]] = {} 

99 files: dict[str, list[tuple[str, str]]] = {} 

100 for name, data in raw_bundles.items(): 

101 data = data or {} 

102 requires[name] = as_list(data.get("requires")) 

103 files[name] = _bundle_file_entries(data.get("files")) 

104 profiles = { 

105 name: as_list((data or {}).get("bundles")) for name, data in raw_profiles.items() 

106 } 

107 return cls(requires=requires, files=files, profiles=profiles) 

108 

109 def _order(self, names: list[str], *, strict: bool) -> list[str]: 

110 """Return *names* plus their ``requires`` dependencies in dependency-first order.""" 

111 order: list[str] = [] 

112 resolved: set[str] = set() 

113 resolving: set[str] = set() 

114 

115 def _collect(name: str) -> None: 

116 if name not in self.requires: 

117 if strict: 

118 raise SyncError(f"Bundle '{name}' does not exist") 

119 return 

120 if name in resolving: 

121 if strict: 

122 raise SyncError(f"Circular dependency detected for bundle '{name}'") 

123 return 

124 if name in resolved: 

125 return 

126 resolving.add(name) 

127 for dependency in self.requires[name]: 

128 _collect(dependency) 

129 resolving.discard(name) 

130 resolved.add(name) 

131 order.append(name) 

132 

133 for name in names: 

134 _collect(name) 

135 return order 

136 

137 def resolve_to_paths(self, names: list[str]) -> list[str]: 

138 """Resolve bundle *names* (and dependencies) to a deduplicated source-path list.""" 

139 paths: list[str] = [] 

140 seen: set[str] = set() 

141 for name in self._order(names, strict=True): 

142 entries = self.files[name] 

143 sources = [source for source, _ in entries] if entries else [f"bundles/{name}/"] 

144 for source in sources: 

145 if source not in seen: 

146 seen.add(source) 

147 paths.append(source) 

148 return paths 

149 

150 def resolve_to_path_map(self, names: list[str]) -> dict[str, str]: 

151 """Return a source->dest map for remapped entries (and dir bundles map to '').""" 

152 resolved = set(self.resolve_to_paths(names)) 

153 path_map: dict[str, str] = {} 

154 for name in self._order(names, strict=False): 

155 entries = self.files[name] 

156 if entries: 

157 for source, dest in entries: 

158 if source in resolved and source != dest: 

159 path_map[source] = dest 

160 else: 

161 path_map[f"bundles/{name}/"] = "" 

162 return path_map 

163 

164 

165def resolve_bundle_names(template: Template, bundles: Bundles) -> list[str]: 

166 """Expand configured profiles to bundle names and merge with explicit templates.""" 

167 if not template.profiles: 

168 return template.templates 

169 names: list[str] = [] 

170 for profile in template.profiles: 

171 if profile not in bundles.profiles: 

172 available = ", ".join(sorted(bundles.profiles)) or "none" 

173 raise SyncError(f"Profile '{profile}' was not found. Available profiles: {available}") 

174 for bundle in bundles.profiles[profile]: 

175 if bundle not in names: 

176 names.append(bundle) 

177 return list(dict.fromkeys(names + template.templates)) 

178 

179 

180# --------------------------------------------------------------------------- 

181# Cloning + snapshot preparation 

182# ---------------------------------------------------------------------------