Extract and install a plugin from a zip file-like object. Args: zip_file: File-like object containing the zip. plugins_dir: Path to the plugins directory. file_name: Name hint for deriving plugin key when the zip has flat contents. allow_overwrite_key: If set, al
(zip_file, plugins_dir, *, file_name="plugin.zip", allow_overwrite_key=None, allow_overwrite=False)
| 155 | |
| 156 | |
| 157 | def _install_plugin_from_zip(zip_file, plugins_dir, *, file_name="plugin.zip", allow_overwrite_key=None, allow_overwrite=False): |
| 158 | """Extract and install a plugin from a zip file-like object. |
| 159 | |
| 160 | Args: |
| 161 | zip_file: File-like object containing the zip. |
| 162 | plugins_dir: Path to the plugins directory. |
| 163 | file_name: Name hint for deriving plugin key when the zip has flat contents. |
| 164 | allow_overwrite_key: If set, allow overwriting this specific plugin directory. |
| 165 | allow_overwrite: If True, allow overwriting any existing plugin with the same key. |
| 166 | |
| 167 | Returns: |
| 168 | dict with "success" bool, and either "plugin_key" on success or "error" on failure. |
| 169 | """ |
| 170 | try: |
| 171 | zf = zipfile.ZipFile(zip_file) |
| 172 | except zipfile.BadZipFile: |
| 173 | return {"success": False, "error": "Invalid zip file"} |
| 174 | |
| 175 | tmp_root = tempfile.mkdtemp(prefix="plugin_import_") |
| 176 | try: |
| 177 | file_members = [m for m in zf.infolist() if not m.is_dir()] |
| 178 | if not file_members: |
| 179 | return {"success": False, "error": "Archive is empty"} |
| 180 | if len(file_members) > MAX_PLUGIN_IMPORT_FILES: |
| 181 | return {"success": False, "error": "Archive has too many files"} |
| 182 | |
| 183 | total_size = 0 |
| 184 | for member in file_members: |
| 185 | total_size += member.file_size |
| 186 | if member.file_size > MAX_PLUGIN_IMPORT_FILE_BYTES: |
| 187 | return {"success": False, "error": "Archive contains a file that is too large"} |
| 188 | if total_size > MAX_PLUGIN_IMPORT_BYTES: |
| 189 | return {"success": False, "error": "Archive is too large"} |
| 190 | |
| 191 | for member in file_members: |
| 192 | name = member.filename |
| 193 | if not name or name.endswith("/"): |
| 194 | continue |
| 195 | norm = os.path.normpath(name) |
| 196 | if norm.startswith("..") or os.path.isabs(norm): |
| 197 | return {"success": False, "error": "Unsafe path in archive"} |
| 198 | dest_path = os.path.join(tmp_root, norm) |
| 199 | os.makedirs(os.path.dirname(dest_path), exist_ok=True) |
| 200 | with zf.open(member, "r") as src, open(dest_path, "wb") as dst: |
| 201 | shutil.copyfileobj(src, dst) |
| 202 | |
| 203 | # Single walk: find candidate plugin dirs AND logo.png simultaneously |
| 204 | candidates = [] |
| 205 | logo_candidates_raw = [] |
| 206 | for dirpath, dirnames, filenames in os.walk(tmp_root): |
| 207 | depth = len(os.path.relpath(dirpath, tmp_root).split(os.sep)) |
| 208 | has_pluginpy = "plugin.py" in filenames |
| 209 | has_init = "__init__.py" in filenames |
| 210 | if has_pluginpy or has_init: |
| 211 | candidates.append((0 if has_pluginpy else 1, depth, dirpath)) |
| 212 | for filename in filenames: |
| 213 | if filename.lower() == "logo.png": |
| 214 | logo_candidates_raw.append((depth, os.path.join(dirpath, filename))) |
no test coverage detected