refactor(plugins): one helper attributes scan blocks to preserved user files

64860c9c20 pasted the same try/except around the security scan in
_install_plugin_core and update_plugin, and plugins_transaction reached
into plugins_cmd_install for the private _preserved_files_note.

_scan_merged_tree now lives next to _scan_plugin_tree/PluginScanBlocked
in plugins_cmd and both sites call it once. Message, scan_result and the
chained cause are unchanged (checked for matched, unmatched, empty and
missing-scan_result cases). _preserved_files_note is typed
(PluginScanBlocked, list[str]) and drops the nested getattr/str() guards
for the module's usual `scan_result.findings if scan_result is not None`
form.
This commit is contained in:
kshitijk4poor
2026-09-26 22:36:50 +05:30
committed by kshitij
parent 7be3b39931
commit 062a6bfc57
3 changed files with 30 additions and 28 deletions

View File

@@ -211,6 +211,33 @@ def _scan_plugin_tree(plugin_dir: Path, identifier: str, *, force: bool, scan_de
return result
def _preserved_files_note(exc: PluginScanBlocked, merged: list[str]) -> str:
"""Scan-block text for a tree that also holds user files preserved from the installed copy."""
preserved = set(merged)
findings = exc.scan_result.findings if exc.scan_result is not None else ()
hits = sorted({f.file for f in findings if f.file in preserved})
if hits:
note = ("These findings come from user files preserved from the installed copy: "
f"{', '.join(hits)}. Move or remove them and retry the update.")
else:
note = "The scanned tree included user files preserved from the installed copy."
return f"{exc}\n\n{note}"
def _scan_merged_tree(plugin_dir: Path, identifier: str, merged: Optional[list[str]], **kwargs):
""":func:`_scan_plugin_tree` for a candidate that may hold carried user files (*merged*).
A block then names the findings that sit in those files, so user data does not read as a
malicious upstream revision; the original block stays chained as the cause.
"""
try:
return _scan_plugin_tree(plugin_dir, identifier, **kwargs)
except PluginScanBlocked as exc:
if not merged:
raise
raise PluginScanBlocked(_preserved_files_note(exc, merged), scan_result=exc.scan_result) from exc
def _plugins_dir() -> Path:
"""Return the user plugins directory, creating it if needed."""
plugins = get_hermes_home() / "plugins"

View File

@@ -269,19 +269,6 @@ def _refuse_unavailable_portable_plugin(plugin_name: str, tree: Path) -> None:
)
def _preserved_files_note(exc: Exception, merged) -> str:
"""Scan-block text for a tree that also holds user files preserved from the installed copy."""
preserved = {str(rel) for rel in merged}
findings = getattr(getattr(exc, "scan_result", None), "findings", None) or []
hits = sorted({f.file for f in findings if f.file in preserved})
if hits:
note = ("These findings come from user files preserved from the installed copy: "
f"{', '.join(hits)}. Move or remove them and retry the update.")
else:
note = "The scanned tree included user files preserved from the installed copy."
return f"{exc}\n\n{note}"
def _install_plugin_core(
identifier: str,
*,
@@ -345,14 +332,8 @@ def _install_plugin_core(
# admits the final bytes.
merged = before_swap(manifest, tmp_target) if before_swap is not None else None
# Scan BEFORE anything is moved into place; raises PluginScanBlocked when blocked.
try:
_pc()._scan_plugin_tree(tmp_target, identifier, force=force, scan_decision_cb=scan_decision_cb,
reviewed_pin=at_reviewed_pin)
except _pc().PluginScanBlocked as exc:
if not merged:
raise
raise _pc().PluginScanBlocked(_preserved_files_note(exc, merged),
scan_result=exc.scan_result) from exc
_pc()._scan_merged_tree(tmp_target, identifier, merged, force=force, scan_decision_cb=scan_decision_cb,
reviewed_pin=at_reviewed_pin)
if not python_deps:
from pm.workspace import enabled_plugin_dirs

View File

@@ -175,13 +175,7 @@ def update_plugin(
raise pc.PluginOperationError(
f"The updated plugin renamed itself to '{installed_name}', but that plugin already exists.")
pc._check_manifest_version(manifest, installed_name)
try:
pc._scan_plugin_tree(staged, source, force=False)
except pc.PluginScanBlocked as exc:
if not merged:
raise
from hermes_cli.plugins_cmd_install import _preserved_files_note
raise pc.PluginScanBlocked(_preserved_files_note(exc, merged), scan_result=exc.scan_result) from exc
pc._scan_merged_tree(staged, source, merged, force=False)
pc._copy_example_files(staged, pc._console())
_refresh_declared_dependencies(target, staged, manifest, interactive=interactive)
if tree_digest(target) != before: