Merge 002ff4fcbd into 0f16801448
This commit is contained in:
commit
7f69e7f992
|
|
@ -560,8 +560,8 @@ class ArchConfigHandler:
|
|||
warn(f'Warning: --debug mode will write certain credentials to {logger.path}!')
|
||||
|
||||
if args.plugin:
|
||||
plugin_path = Path(args.plugin)
|
||||
load_plugin(plugin_path)
|
||||
# pathlib collapses "https://..." to "https:/..." which breaks URL loading (#3021).
|
||||
load_plugin(args.plugin)
|
||||
|
||||
if args.creds_decryption_key is None:
|
||||
if os.environ.get('ARCHINSTALL_CREDS_DECRYPTION_KEY'):
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ import hashlib
|
|||
import importlib.util
|
||||
import os
|
||||
import sys
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from importlib import metadata
|
||||
|
|
@ -34,21 +35,42 @@ def plugin(f, *args, **kwargs) -> None: # type: ignore[no-untyped-def]
|
|||
plugins[f.__name__] = f
|
||||
|
||||
|
||||
def _localize_path(path: Path) -> Path:
|
||||
def _localize_path(path: str | Path) -> Path:
|
||||
"""
|
||||
Support structures for load_plugin()
|
||||
"""
|
||||
url = urllib.parse.urlparse(str(path))
|
||||
# Keep as string to preserve URL format (Path normalization breaks URLs)
|
||||
path_str = str(path)
|
||||
url = urllib.parse.urlparse(path_str)
|
||||
|
||||
if url.scheme and url.scheme in ('https', 'http'):
|
||||
converted_path = Path(f'/tmp/{path.stem}_{hashlib.md5(os.urandom(12)).hexdigest()}.py')
|
||||
if url.scheme == 'http':
|
||||
error(f'Insecure HTTP URL {path_str} is not allowed for downloading plugins. Please use HTTPS.')
|
||||
raise ValueError('Insecure HTTP URLs are blocked for security reasons.')
|
||||
|
||||
with open(converted_path, 'w') as temp_file:
|
||||
temp_file.write(urllib.request.urlopen(url.geturl()).read().decode('utf-8'))
|
||||
# Extract filename from the URL path component
|
||||
# Use os.path.basename instead of path.stem to handle URLs correctly
|
||||
url_path = url.path
|
||||
filename = os.path.basename(url_path) if url_path else 'plugin'
|
||||
# Remove .py extension if present for the temporary filename format
|
||||
if filename.endswith('.py'):
|
||||
filename_base = filename.replace('.py', '')
|
||||
else:
|
||||
filename_base = filename
|
||||
|
||||
converted_path = Path(f'/tmp/{filename_base}_{hashlib.md5(os.urandom(12)).hexdigest()}.py')
|
||||
|
||||
with open(converted_path, 'wb') as temp_file:
|
||||
try:
|
||||
with urllib.request.urlopen(path_str, timeout=15) as response:
|
||||
temp_file.write(response.read())
|
||||
except urllib.error.URLError as e:
|
||||
error(f'Failed to download plugin from {path_str}: {e}')
|
||||
raise
|
||||
|
||||
return converted_path
|
||||
else:
|
||||
return path
|
||||
return Path(path)
|
||||
|
||||
|
||||
def _import_via_path(path: Path, namespace: str | None = None) -> str:
|
||||
|
|
@ -60,10 +82,16 @@ def _import_via_path(path: Path, namespace: str | None = None) -> str:
|
|||
|
||||
try:
|
||||
spec = importlib.util.spec_from_file_location(namespace, path)
|
||||
if spec and spec.loader:
|
||||
imported = importlib.util.module_from_spec(spec)
|
||||
sys.modules[namespace] = imported
|
||||
spec.loader.exec_module(sys.modules[namespace])
|
||||
if spec is None or spec.loader is None:
|
||||
error(
|
||||
f'Could not load plugin module spec from {path}',
|
||||
f'The above error was detected when loading the plugin: {path}',
|
||||
)
|
||||
return ''
|
||||
|
||||
imported = importlib.util.module_from_spec(spec)
|
||||
sys.modules[namespace] = imported
|
||||
spec.loader.exec_module(imported)
|
||||
|
||||
return namespace
|
||||
except Exception as err:
|
||||
|
|
@ -77,21 +105,23 @@ def _import_via_path(path: Path, namespace: str | None = None) -> str:
|
|||
except Exception:
|
||||
pass
|
||||
|
||||
return namespace
|
||||
return ''
|
||||
|
||||
|
||||
def load_plugin(path: Path) -> None:
|
||||
def load_plugin(path: str | Path) -> None:
|
||||
namespace: str | None = None
|
||||
parsed_url = urllib.parse.urlparse(str(path))
|
||||
# Keep URL as string to preserve scheme (avoid Path normalization)
|
||||
path_str = str(path) if isinstance(path, Path) else path
|
||||
parsed_url = urllib.parse.urlparse(path_str)
|
||||
info(f'Loading plugin from url {parsed_url}')
|
||||
|
||||
# The Profile was not a direct match on a remote URL
|
||||
if not parsed_url.scheme:
|
||||
# Path was not found in any known examples, check if it's an absolute path
|
||||
if os.path.isfile(path):
|
||||
namespace = _import_via_path(path)
|
||||
if os.path.isfile(path_str):
|
||||
namespace = _import_via_path(Path(path_str))
|
||||
elif parsed_url.scheme in ('https', 'http'):
|
||||
localized = _localize_path(path)
|
||||
localized = _localize_path(path_str)
|
||||
namespace = _import_via_path(localized)
|
||||
|
||||
if namespace and namespace in sys.modules:
|
||||
|
|
@ -102,7 +132,15 @@ def load_plugin(path: Path) -> None:
|
|||
if version is not None:
|
||||
version_major_and_minor = version.rsplit('.', 1)[0]
|
||||
|
||||
if sys.modules[namespace].__archinstall__version__ < float(version_major_and_minor):
|
||||
plugin_version_raw = getattr(sys.modules[namespace], '__archinstall__version__', '0.0')
|
||||
|
||||
def parse_version(v: str | float) -> tuple[int, ...]:
|
||||
return tuple(int(x) for x in str(v).split('.') if x.isdigit())
|
||||
|
||||
plugin_version = parse_version(plugin_version_raw)
|
||||
system_version = parse_version(version_major_and_minor)
|
||||
|
||||
if plugin_version and system_version and plugin_version < system_version:
|
||||
error(f'Plugin {sys.modules[namespace]} does not support the current Archinstall version.')
|
||||
|
||||
# Locate the plugin entry-point called Plugin()
|
||||
|
|
|
|||
|
|
@ -0,0 +1,38 @@
|
|||
import urllib.parse
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from pytest import MonkeyPatch
|
||||
|
||||
from archinstall.lib.args import ArchConfigHandler
|
||||
|
||||
|
||||
def test_path_corrupts_https_url_authority_issue_3021() -> None:
|
||||
"""pathlib.Path is not safe for URL strings: POSIX normalization drops one slash after the scheme."""
|
||||
url = 'https://raw.githubusercontent.com/phisch/archinstall-aur/refs/heads/master/archinstall-aur.py'
|
||||
broken = urllib.parse.urlparse(str(Path(url)))
|
||||
assert broken.netloc == ''
|
||||
assert broken.scheme == 'https'
|
||||
|
||||
|
||||
def test_cli_https_plugin_passes_unparsed_string_to_load_plugin(monkeypatch: MonkeyPatch) -> None:
|
||||
url = 'https://raw.githubusercontent.com/phisch/archinstall-aur/refs/heads/master/archinstall-aur.py'
|
||||
received: list[object] = []
|
||||
|
||||
def capture(path: object) -> None:
|
||||
received.append(path)
|
||||
|
||||
monkeypatch.setattr('archinstall.lib.args.load_plugin', capture)
|
||||
monkeypatch.setattr('sys.argv', ['archinstall', '--plugin', url])
|
||||
ArchConfigHandler()
|
||||
assert len(received) == 1
|
||||
parsed = urllib.parse.urlparse(str(received[0]))
|
||||
assert parsed.scheme == 'https'
|
||||
assert parsed.netloc == 'raw.githubusercontent.com'
|
||||
|
||||
|
||||
def test_localize_path_rejects_http() -> None:
|
||||
from archinstall.lib.plugins import _localize_path
|
||||
|
||||
with pytest.raises(ValueError, match='Insecure HTTP'):
|
||||
_localize_path('http://example.com/plugin.py')
|
||||
Loading…
Reference in New Issue