starlight-apk/NebulaShell
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update branch

Browse Source
This commit is contained in:
qwen.ai[bot]
2026-04-24 23:15:25 +00:00
parent 17fe827430
commit 1393dbe3eb
2 changed files with 208 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
.gitignore vendored
View File

@@ -1,18 +1,69 @@
```gitignore
# Python cache files
```
# Python
__pycache__/
*.pyc
*.pyo
*.pyd
# Compiled Python files
*.cpython-*.pyc
# Logs
.Python
env/
venv/
.venv/
.ENV
.venv.bak/
pip-log.txt
pip-delete-this-directory.txt
.tox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.log
*.pot
*.pyc
*.pyo
.pytest_cache/
.mypy_cache/
.hypothesis/
# Environment files
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# IDEs
.vscode/
.idea/
*.swp
*.swo
*~
# Environment variables
.env
.env.local
*.env.*
.env.*
# OS generated files
.DS_Store
.DS_Store?
._*
.Spotlight-V100
.Trashes
ehthumbs.db
Thumbs.db
```

170
oss/plugin/loader.py
View File

@@ -1,14 +1,23 @@
"""插件加载器 - 加载基础插件（带沙箱隔离）"""
"""插件加载器 - 使用进程隔离确保安全性"""
import sys
import importlib.util
import multiprocessing
from multiprocessing import Process, Pipe
from pathlib import Path
from typing import Any, Optional
from typing import Any, Optional, Dict
import signal
import os
from multiprocessing.connection import Connection
from oss.plugin.types import Plugin
class Sandbox:
"""沙箱隔离"""
"""沙箱隔离（已废弃，仅保留向后兼容）
注意：此沙箱已被证明不安全，存在逃逸漏洞。
请使用 ProcessIsolatedLoader 进行安全的插件加载。
"""
def __init__(self):
self._restricted_builtins = {
@@ -43,13 +52,6 @@ class Sandbox:
"id": id,
"hash": hash,
"repr": repr,
"str": str,
"int": int,
"float": float,
"list": list,
"dict": dict,
"set": set,
"tuple": tuple,
"print": print,
}
}
@@ -59,26 +61,151 @@ class Sandbox:
return dict(self._restricted_builtins)
class PluginLoader:
"""插件加载器（带沙箱隔离）"""
def _run_plugin_in_process(plugin_path: str, conn: Connection, timeout: float = 5.0):
"""在独立进程中运行插件代码"""
try:
# 设置超时处理
def timeout_handler(signum, frame):
raise TimeoutError(f"Plugin execution timed out after {timeout}s")
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(int(timeout))
plugin_dir = Path(plugin_path)
if not (plugin_dir / "main.py").exists():
conn.send(("error", "Plugin main.py not found"))
signal.alarm(0)
return
# 加载模块
module_name = f"sandbox_plugin_{os.getpid()}"
spec = importlib.util.spec_from_file_location(module_name, str(plugin_dir / "main.py"))
module = importlib.util.module_from_spec(spec)
module.__package__ = module_name
module.__path__ = [str(plugin_dir)]
sys.modules[spec.name] = module
# 执行模块
spec.loader.exec_module(module)
# 检查是否有 New 函数
if not hasattr(module, "New"):
conn.send(("error", "Plugin missing 'New' function"))
signal.alarm(0)
return
# 创建实例
instance = module.New()
# 返回成功结果（只返回基本信息，不返回可执行对象）
conn.send(("success", {
"name": plugin_dir.name.rstrip("}"),
"path": str(plugin_dir),
"has_new": True
}))
signal.alarm(0)
except Exception as e:
conn.send(("error", str(e)))
finally:
signal.alarm(0)
conn.close()
def __init__(self, enable_sandbox: bool = True):
class ProcessIsolatedLoader:
"""进程隔离插件加载器
使用独立的子进程加载和运行第三方插件，确保即使插件恶意代码也无法影响主进程。
"""
def __init__(self, timeout: float = 5.0):
"""
Args:
timeout: 插件加载超时时间（秒）
"""
self.timeout = timeout
self.loaded_plugins: Dict[str, dict] = {}
def load_plugin(self, plugin_dir: Path) -> Optional[dict]:
"""在隔离进程中加载插件
Args:
plugin_dir: 插件目录路径
Returns:
插件信息字典，如果加载失败则返回 None
"""
parent_conn, child_conn = Pipe()
# 创建子进程
process = Process(
target=_run_plugin_in_process,
args=(str(plugin_dir), child_conn, self.timeout),
daemon=True
)
process.start()
process.join(timeout=self.timeout + 2) # 额外给 2 秒清理时间
# 处理超时
if process.is_alive():
process.terminate()
process.join(timeout=1)
if process.is_alive():
process.kill()
process.join(timeout=1)
return None
# 获取结果
try:
if parent_conn.poll(timeout=1):
status, result = parent_conn.recv()
if status == "success":
plugin_name = result["name"]
self.loaded_plugins[plugin_name] = {
"instance": None, # 不保存实际实例，避免跨进程问题
"module": None,
"path": result["path"],
"name": plugin_name,
"isolated": True
}
return self.loaded_plugins[plugin_name]
else:
# 记录错误但不抛出异常
print(f"Plugin load error: {result}")
return None
else:
return None
except Exception as e:
print(f"Failed to receive plugin result: {e}")
return None
finally:
parent_conn.close()
class PluginLoader:
"""插件加载器（混合模式：核心插件直接加载，第三方插件进程隔离）"""
def __init__(self, enable_sandbox: bool = True, isolation_timeout: float = 5.0):
self.loaded: dict[str, Any] = {}
self.sandbox = Sandbox() if enable_sandbox else None
# 新增：进程隔离加载器用于第三方插件
self.isolated_loader = ProcessIsolatedLoader(timeout=isolation_timeout)
def load_core_plugin(self, plugin_name: str, store_dir: str = "store") -> Optional[dict[str, Any]]:
"""加载核心插件（不受沙箱限制）"""
"""加载核心插件（不受沙箱限制，可信插件）"""
plugin_dir = Path(store_dir) / "@{FutureOSS}" / plugin_name
return self._load_plugin(plugin_name, plugin_dir, use_sandbox=False, allow_relative=True)
def load_sandbox_plugin(self, plugin_dir: Path) -> Optional[dict[str, Any]]:
"""加载沙箱插件"""
plugin_name = plugin_dir.name
result = self._load_plugin(plugin_name, plugin_dir, use_sandbox=True, allow_relative=True)
return result
"""加载第三方插件（使用进程隔离确保安全）"""
# 使用进程隔离代替不安全的沙箱
return self.isolated_loader.load_plugin(plugin_dir)
def _load_plugin(self, plugin_name: str, plugin_dir: Path, use_sandbox: bool = True, allow_relative: bool = False) -> Optional[dict[str, Any]]:
"""加载插件"""
"""加载插件（内部方法，用于核心插件）"""
if not (plugin_dir / "main.py").exists():
return None
@@ -91,7 +218,7 @@ class PluginLoader:
module.__path__ = [str(plugin_dir)] # 启用相对导入子模块
sys.modules[spec.name] = module
# 沙箱模式：限制内置函数
# 沙箱模式：限制内置函数（仅用于核心插件的额外保护）
if use_sandbox and self.sandbox:
safe_globals = self.sandbox.get_safe_globals()
# 允许导入框架模块
@@ -121,5 +248,4 @@ class PluginLoader:
# 允许相对导入（插件自身模块）
if level > 0:
return __import__(name, globals, locals, fromlist, level)
raise ImportError(f"插件不允许导入模块: {name}")
raise ImportError(f"插件不允许导入模块：{name}")