fix(plugin): support flat plugin upload archives

[p1930n]

Fixes #7604

Modifications / 改动点

• Support plugin upload archives that place files directly at the zip root.

• Preserve existing behavior for archives with a single top-level directory.

• Avoid flattening conflicts when the archive root contains a same-named child directory.

• Reject unsafe archive member paths before extraction.

• Report empty plugin archives with a clear error.

• This is NOT a breaking change. / 这不是一个破坏性变更。

Screenshots or Test Results / 运行截图或测试结果

python -m uv run pytest tests/test_updator_socks.py -q
19 passed

python -m uv run ruff format . --check
418 files already formatted

python -m uv run ruff check .
All checks passed!

---

Checklist / 检查清单

• No new feature is added.
• My changes have been well-tested, and verification steps are provided above.
• No new dependencies are introduced.
• My changes do not introduce malicious code.

[gemini-code-assist]

Code Review

This pull request refactors the plugin update mechanism to use pathlib, enhances security by implementing safe extraction checks against path traversal, and introduces logic to automatically flatten archives containing a single root directory. Feedback highlights a potential bug in the flattening logic where conflicting directory names could lead to premature deletion during iteration, and recommends catching Exception instead of BaseException to adhere to PEP 8 standards.

[sourcery-ai]

Hey - I've found 3 issues, and left some high level feedback:

• The temporary renaming in _rename_extracted_root_for_flattening (.<name>.tmp) will delete any pre-existing directory/file with that name under target_path; consider generating a unique temp directory name (e.g., with a random suffix) to avoid unintentionally removing user data.
• _get_archive_root_dir and _extract_archive_safely both call _get_safe_member_parts on every member; you could refactor to compute and reuse the parsed member parts once per archive to reduce duplication and keep all safety/structure logic in one place.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- The temporary renaming in `_rename_extracted_root_for_flattening` (`.<name>.tmp`) will delete any pre-existing directory/file with that name under `target_path`; consider generating a unique temp directory name (e.g., with a random suffix) to avoid unintentionally removing user data.
- `_get_archive_root_dir` and `_extract_archive_safely` both call `_get_safe_member_parts` on every member; you could refactor to compute and reuse the parsed member parts once per archive to reduce duplication and keep all safety/structure logic in one place.

## Individual Comments

### Comment 1
<location path="tests/test_updator_socks.py" line_range="181-190" />
<code_context>
     assert calls["unzip"] == (str(expected_path) + ".zip", str(expected_path))


+def test_plugin_updator_unzip_file_accepts_flat_plugin_archive(tmp_path: Path) -> None:
+    archive_path = tmp_path / "flat_plugin.zip"
+    target_path = tmp_path / "plugin_upload"
+    with zipfile.ZipFile(archive_path, "w") as archive:
+        archive.writestr("main.py", "print('loaded')\n")
+        archive.writestr("metadata.yaml", "name: flat_plugin\n")
+        archive.writestr("commands/__init__.py", "")
+
+    PluginUpdator().unzip_file(str(archive_path), str(target_path))
+
+    assert (target_path / "main.py").read_text(encoding="utf-8") == "print('loaded')\n"
+    assert (target_path / "metadata.yaml").read_text(encoding="utf-8") == (
+        "name: flat_plugin\n"
+    )
+    assert (target_path / "commands" / "__init__.py").exists()
+    assert not archive_path.exists()
+
+
</code_context>
<issue_to_address>
**suggestion (testing):** Add a test to cover archives with multiple top-level entries to ensure they are not flattened incorrectly.

The current tests cover only a fully flat archive and a single-root-dir archive. Please also add a test for a mixed archive, e.g. `foo/main.py` and `bar/other.py`, where `_get_archive_root_dir` should return `None` and no flattening should occur. The test should verify that:

- Files are extracted under their respective top-level directories; and
- No temporary rename/flattening happens (both `foo` and `bar` remain as directories under `target_path`).

This will protect against regressions in the single-root vs multi-root detection logic.
</issue_to_address>

### Comment 2
<location path="tests/test_updator_socks.py" line_range="233-244" />
<code_context>
+    assert not archive_path.exists()
+
+
+def test_plugin_updator_unzip_file_rejects_unsafe_member_path(
+    tmp_path: Path,
+) -> None:
+    archive_path = tmp_path / "unsafe_plugin.zip"
+    target_path = tmp_path / "plugin_upload"
+    with zipfile.ZipFile(archive_path, "w") as archive:
+        archive.writestr("../escape.py", "print('escape')\n")
+
+    with pytest.raises(ValueError, match="Unsafe path in zip archive"):
+        PluginUpdator().unzip_file(str(archive_path), str(target_path))
+
+    assert not (tmp_path / "escape.py").exists()
+
+
</code_context>
<issue_to_address>
**suggestion (testing):** Extend unsafe-path tests to cover additional path variants like backslashes, absolute paths, and colon-containing segments.

Since `_get_safe_member_parts` also rejects backslashes, absolute paths, `.`/`..` segments, and names containing `:`, please add a few targeted cases, e.g.:

- `"..\\escape.py"` and `"foo\\bar.py"` (Windows-style separators)
- An absolute path like `"/etc/passwd"`
- A segment with a colon such as `"C:drive/file.py"`

Each should assert that `PluginUpdator().unzip_file(...)` raises `ValueError` and that no files are written under or outside `target_path`.

```suggestion
def test_plugin_updator_unzip_file_rejects_unsafe_member_path(
    tmp_path: Path,
) -> None:
    target_path = tmp_path / "plugin_upload"

    unsafe_members = [
        "../escape.py",        # parent-directory escape
        "..\\escape.py",       # Windows-style parent-directory escape
        "foo\\bar.py",         # Windows-style subdir separator
        "/etc/passwd",         # absolute path
        "C:drive/file.py",     # colon-containing segment
    ]

    for index, member in enumerate(unsafe_members):
        archive_path = tmp_path / f"unsafe_plugin_{index}.zip"
        with zipfile.ZipFile(archive_path, "w") as archive:
            archive.writestr(member, "print('escape')\n")

        with pytest.raises(ValueError, match="Unsafe path in zip archive"):
            PluginUpdator().unzip_file(str(archive_path), str(target_path))

        # No files should be created under the target path for any unsafe member.
        if target_path.exists():
            assert not any(target_path.rglob("*"))

    # For the ../escape.py case, ensure nothing escapes the target directory.
    assert not (tmp_path / "escape.py").exists()
```
</issue_to_address>

### Comment 3
<location path="astrbot/core/star/updator.py" line_range="74" />
<code_context>
         return plugin_path

     def unzip_file(self, zip_path: str, target_dir: str) -> None:
-        ensure_dir(target_dir)
-        update_dir = ""
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying `unzip_file` by inlining validation, extraction, and flattening logic and removing now-unnecessary helper methods to make the control flow clearer.

You can keep the safety/flattening behavior while simplifying control flow and reducing indirection.

### 1. Merge validation and extraction into a single loop

`_extract_archive_safely` currently walks `archive.infolist()` twice and delegates only to `_get_safe_member_parts`. You can simplify by validating and extracting in one pass and inlining the logic in `unzip_file` (or keeping a single small helper):

```python
def unzip_file(self, zip_path: str, target_dir: str) -> None:
    target_path = Path(target_dir)
    ensure_dir(target_path)
    logger.info(f"Extracting archive: {zip_path}")

    archive_root_dir = None
    with zipfile.ZipFile(zip_path, "r") as z:
        archive_root_dir = self._get_archive_root_dir(z.infolist())
        for member in z.infolist():
            # validate path before extraction
            self._get_safe_member_parts(member.filename)
            z.extract(member, target_path)
    ...
```

With this, `_extract_archive_safely` becomes unnecessary, and the main flow is easier to follow (no extra classmethod + double traversal).

### 2. Flatten without temp rename and helper

You can flatten directly from the extracted root directory without renaming to a temp directory, and inline the removal logic instead of `_rename_extracted_root_for_flattening` + `_remove_existing_path`:

```python
extracted_root = target_path / archive_root_dir if archive_root_dir else None
if extracted_root and extracted_root.is_dir():
    for child in list(extracted_root.iterdir()):
        destination = target_path / child.name
        if destination.is_dir() and not destination.is_symlink():
            shutil.rmtree(destination, onerror=on_error)
        elif destination.exists() or destination.is_symlink():
            destination.unlink()
        shutil.move(str(child), str(target_path))
```

Then the cleanup can simply be:

```python
try:
    logger.info(
        f"Removing temporary files: {zip_path}"
        + (f" and {extracted_root}" if extracted_root else "")
    )
    if extracted_root and extracted_root.exists():
        shutil.rmtree(extracted_root, onerror=on_error)
    os.remove(zip_path)
except Exception:
    ...
```

This removes `_rename_extracted_root_for_flattening` and `_remove_existing_path`, keeps the same behavior (including directory vs symlink handling), and makes the flattening behavior visible directly in `unzip_file`.

### 3. Make helper signatures consistent (optional)

Since `_get_archive_root_dir` and `_get_safe_member_parts` don’t need `cls`, they can both be `@staticmethod`s to avoid mixing `@classmethod` and `@staticmethod` for tightly related helpers:

```python
@staticmethod
def _get_archive_root_dir(members: list[zipfile.ZipInfo]) -> str | None:
    root_dir = None
    for member in members:
        parts = PluginUpdator._get_safe_member_parts(member.filename)
        ...
```

This reduces one more axis of indirection without changing functionality.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[p1930n]

我在新commit 99ae0849 处理了

_rename_extracted_root_for_flattening() 里已经不再使用固定的 .<name>.tmp了，改成了带 UUID 的唯一临时路径，避免误删 target_path 下已有的同名文件或目录。

另外也补了两个测试：

1. multi-root archive，例如 plugin_a/main.py + plugin_b/main.py，确认不会被 flatten；
2. unsafe path variants，覆盖 parent traversal、absolute path、colon-containing segment，以及 backslash path。

我下一步准备补齐空zip和根部空目录边界处理

[p1930n]

我在 2d5509e 继续补了两个边界：

1. 空 zip / 只有目录没有文件的 zip 现在会直接丢出 ValueError("Empty plugin archive")，避免后续出现不清晰的间接错误
2. 补了 plugin/main.py + docs/ 这种根部额外空目录的测试，并让这类包不再被误判成单根目录 flatten