RomanNum3ral
/
archinstall
mirror of https://github.com/archlinux/archinstall.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix(plugins): restore args parsing and complete #3021 fixes 

Pass plugin paths as strings so HTTPS URLs are not mangled by pathlib.
Fix mixed tab/space indentation in ArchConfig.from_config that broke
imports. Harden plugin import (spec handling, failed-import cleanup),
and skip version comparison when version tuples are empty. Add tests
for URL handling and HTTP rejection.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
hariomphulre 2026-05-15 02:06:25 +05:30
parent ef7d71bb7d
commit 002ff4fcbd
3 changed files with 230 additions and 191 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

170
archinstall/lib/args.py
View File

@ -37,26 +37,26 @@ from archinstall.tui.components import tui
@p_dataclass
class Arguments:
config: Path | None = None
config_url: str | None = None
creds: Path | None = None
creds_url: str | None = None
creds_decryption_key: str | None = None
silent: bool = False
dry_run: bool = False
script: str | None = None
mountpoint: Path = Path('/mnt')
skip_ntp: bool = False
skip_wkd: bool = False
skip_boot: bool = False
debug: bool = False
offline: bool = False
no_pkg_lookups: bool = False
plugin: str | None = None
skip_version_check: bool = False
skip_wifi_check: bool = False
advanced: bool = False
verbose: bool = False
config: Path | None = None
config_url: str | None = None
creds: Path | None = None
creds_url: str | None = None
creds_decryption_key: str | None = None
silent: bool = False
dry_run: bool = False
script: str | None = None
mountpoint: Path = Path('/mnt')
skip_ntp: bool = False
skip_wkd: bool = False
skip_boot: bool = False
debug: bool = False
offline: bool = False
no_pkg_lookups: bool = False
plugin: str | None = None
skip_version_check: bool = False
skip_wifi_check: bool = False
advanced: bool = False
verbose: bool = False
class ArchConfigType(StrEnum):
@ -276,90 +276,90 @@ class ArchConfig:
if additional_repositories := args_config.get('additional-repositories', []):
backwards_compatible_repo = [Repository(r) for r in additional_repositories]
arch_config.mirror_config = MirrorConfiguration.parse_args(
mirror_config,
backwards_compatible_repo,
)
arch_config.mirror_config = MirrorConfiguration.parse_args(
mirror_config,
backwards_compatible_repo,
)
if net_config := args_config.get('network_config', None):
arch_config.network_config = NetworkConfiguration.parse_arg(net_config)
if net_config := args_config.get('network_config', None):
arch_config.network_config = NetworkConfiguration.parse_arg(net_config)
if bootloader_config_dict := args_config.get('bootloader_config', None):
arch_config.bootloader_config = BootloaderConfiguration.parse_arg(bootloader_config_dict, args.skip_boot)
# DEPRECATED: separate bootloader and uki fields (backward compatibility)
elif bootloader_str := args_config.get('bootloader', None):
bootloader = Bootloader.from_arg(bootloader_str, args.skip_boot)
uki = args_config.get('uki', False)
if uki and not bootloader.has_uki_support():
uki = False
arch_config.bootloader_config = BootloaderConfiguration(bootloader=bootloader, uki=uki, removable=True)
if bootloader_config_dict := args_config.get('bootloader_config', None):
arch_config.bootloader_config = BootloaderConfiguration.parse_arg(bootloader_config_dict, args.skip_boot)
# DEPRECATED: separate bootloader and uki fields (backward compatibility)
elif bootloader_str := args_config.get('bootloader', None):
bootloader = Bootloader.from_arg(bootloader_str, args.skip_boot)
uki = args_config.get('uki', False)
if uki and not bootloader.has_uki_support():
uki = False
arch_config.bootloader_config = BootloaderConfiguration(bootloader=bootloader, uki=uki, removable=True)
# deprecated: backwards compatibility
audio_config_args = args_config.get('audio_config', None)
app_config_args = args_config.get('app_config', None)
# deprecated: backwards compatibility
audio_config_args = args_config.get('audio_config', None)
app_config_args = args_config.get('app_config', None)
if audio_config_args is not None or app_config_args is not None:
arch_config.app_config = ApplicationConfiguration.parse_arg(app_config_args, audio_config_args)
if audio_config_args is not None or app_config_args is not None:
arch_config.app_config = ApplicationConfiguration.parse_arg(app_config_args, audio_config_args)
if auth_config_args := args_config.get('auth_config', None):
arch_config.auth_config = AuthenticationConfiguration.parse_arg(auth_config_args)
if auth_config_args := args_config.get('auth_config', None):
arch_config.auth_config = AuthenticationConfiguration.parse_arg(auth_config_args)
if hostname := args_config.get('hostname', ''):
arch_config.hostname = hostname
if hostname := args_config.get('hostname', ''):
arch_config.hostname = hostname
if kernels := args_config.get('kernels', []):
arch_config.kernels = kernels
if kernels := args_config.get('kernels', []):
arch_config.kernels = kernels
arch_config.ntp = args_config.get('ntp', True)
arch_config.ntp = args_config.get('ntp', True)
if packages := args_config.get('packages', []):
arch_config.packages = packages
if packages := args_config.get('packages', []):
arch_config.packages = packages
if pacman_config := args_config.get('pacman_config', None):
arch_config.pacman_config = PacmanConfiguration.parse_arg(pacman_config)
elif parallel_downloads := args_config.get('parallel_downloads', 0):
arch_config.pacman_config = PacmanConfiguration(parallel_downloads=int(parallel_downloads))
if pacman_config := args_config.get('pacman_config', None):
arch_config.pacman_config = PacmanConfiguration.parse_arg(pacman_config)
elif parallel_downloads := args_config.get('parallel_downloads', 0):
arch_config.pacman_config = PacmanConfiguration(parallel_downloads=int(parallel_downloads))
swap_arg = args_config.get('swap')
if swap_arg is not None:
arch_config.swap = ZramConfiguration.parse_arg(swap_arg)
swap_arg = args_config.get('swap')
if swap_arg is not None:
arch_config.swap = ZramConfiguration.parse_arg(swap_arg)
if timezone := args_config.get('timezone', 'UTC'):
arch_config.timezone = timezone
if timezone := args_config.get('timezone', 'UTC'):
arch_config.timezone = timezone
if services := args_config.get('services', []):
arch_config.services = services
if services := args_config.get('services', []):
arch_config.services = services
# DEPRECATED: backwards compatibility
root_password = None
if root_password := args_config.get('!root-password', None):
root_password = Password(plaintext=root_password)
# DEPRECATED: backwards compatibility
root_password = None
if root_password := args_config.get('!root-password', None):
root_password = Password(plaintext=root_password)
if enc_password := args_config.get('root_enc_password', None):
root_password = Password(enc_password=enc_password)
if enc_password := args_config.get('root_enc_password', None):
root_password = Password(enc_password=enc_password)
if root_password is not None:
if arch_config.auth_config is None:
arch_config.auth_config = AuthenticationConfiguration()
arch_config.auth_config.root_enc_password = root_password
if root_password is not None:
if arch_config.auth_config is None:
arch_config.auth_config = AuthenticationConfiguration()
arch_config.auth_config.root_enc_password = root_password
# DEPRECATED: backwards compatibility
users: list[User] = []
if args_users := args_config.get('!users', None):
users = User.parse_arguments(args_users)
# DEPRECATED: backwards compatibility
users: list[User] = []
if args_users := args_config.get('!users', None):
users = User.parse_arguments(args_users)
if args_users := args_config.get('users', None):
users = User.parse_arguments(args_users)
if args_users := args_config.get('users', None):
users = User.parse_arguments(args_users)
if users:
if arch_config.auth_config is None:
arch_config.auth_config = AuthenticationConfiguration()
arch_config.auth_config.users = users
if users:
if arch_config.auth_config is None:
arch_config.auth_config = AuthenticationConfiguration()
arch_config.auth_config.users = users
if custom_commands := args_config.get('custom_commands', []):
arch_config.custom_commands = custom_commands
if custom_commands := args_config.get('custom_commands', []):
arch_config.custom_commands = custom_commands
return arch_config
return arch_config
class ArchConfigHandler:
@ -549,8 +549,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'):

213
archinstall/lib/plugins.py
View File

@ -18,140 +18,141 @@ plugins = {}
# 2: Load the plugin entrypoint
# 3: Initiate the plugin and store it as .name in plugins
for plugin_definition in metadata.entry_points().select(group='archinstall.plugin'):
plugin_entrypoint = plugin_definition.load()
plugin_entrypoint = plugin_definition.load()
try:
plugins[plugin_definition.name] = plugin_entrypoint()
except Exception as err:
error(
f'Error: {err}',
f'The above error was detected when loading the plugin: {plugin_definition}',
)
try:
plugins[plugin_definition.name] = plugin_entrypoint()
except Exception as err:
error(
f'Error: {err}',
f'The above error was detected when loading the plugin: {plugin_definition}',
)
# @archinstall.plugin decorator hook to programmatically add
# plugins in runtime. Useful in profiles_bck and other things.
def plugin(f, *args, **kwargs) -> None: # type: ignore[no-untyped-def]
plugins[f.__name__] = f
plugins[f.__name__] = f
def _localize_path(path: str | Path) -> Path:
"""
Support structures for load_plugin()
"""
# Keep as string to preserve URL format (Path normalization breaks URLs)
path_str = str(path)
url = urllib.parse.urlparse(path_str)
"""
Support structures for load_plugin()
"""
# 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'):
# FIXED: Prevent arbitrary code execution over unencrypted HTTP
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.')
if url.scheme and url.scheme in ('https', 'http'):
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.')
# 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')
# 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
# FIXED: Open in 'wb' (write-binary) mode to safely write downloaded bytes without assuming UTF-8
with open(converted_path, 'wb') as temp_file:
try:
# FIXED: Added a 15-second timeout and wrapped urlopen in a `with` statement to close the socket cleanly
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
converted_path = Path(f'/tmp/{filename_base}_{hashlib.md5(os.urandom(12)).hexdigest()}.py')
return converted_path
else:
return Path(path)
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(path)
def _import_via_path(path: Path, namespace: str | None = None) -> str:
if not namespace:
namespace = os.path.basename(path)
if not namespace:
namespace = os.path.basename(path)
if namespace == '__init__.py':
namespace = path.parent.name
if namespace == '__init__.py':
namespace = path.parent.name
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])
try:
spec = importlib.util.spec_from_file_location(namespace, path)
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 ''
return namespace
except Exception as err:
error(
f'Error: {err}',
f'The above error was detected when loading the plugin: {path}',
)
imported = importlib.util.module_from_spec(spec)
sys.modules[namespace] = imported
spec.loader.exec_module(imported)
try:
del sys.modules[namespace]
except Exception:
pass
return namespace
except Exception as err:
error(
f'Error: {err}',
f'The above error was detected when loading the plugin: {path}',
)
return namespace
try:
del sys.modules[namespace]
except Exception:
pass
return ''
def load_plugin(path: str | Path) -> None:
namespace: str | None = None
# 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}')
namespace: str | None = None
# 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_str):
namespace = _import_via_path(Path(path_str))
elif parsed_url.scheme in ('https', 'http'):
localized = _localize_path(path_str)
namespace = _import_via_path(localized)
# 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_str):
namespace = _import_via_path(Path(path_str))
elif parsed_url.scheme in ('https', 'http'):
localized = _localize_path(path_str)
namespace = _import_via_path(localized)
if namespace and namespace in sys.modules:
# Version dependency via __archinstall__version__ variable (if present) in the plugin
# Any errors in version inconsistency will be handled through normal error handling if not defined.
version = get_version()
if namespace and namespace in sys.modules:
# Version dependency via __archinstall__version__ variable (if present) in the plugin
# Any errors in version inconsistency will be handled through normal error handling if not defined.
version = get_version()
if version is not None:
version_major_and_minor = version.rsplit('.', 1)[0]
# FIXED: Safely fetch the plugin version attribute, defaulting to "0.0" if missing
plugin_version_raw = getattr(sys.modules[namespace], '__archinstall__version__', '0.0')
if version is not None:
version_major_and_minor = version.rsplit('.', 1)[0]
# FIXED: Safely parse versions into a tuple for integer comparison, preventing the float("2.10") == 2.1 bug
def parse_version(v: str | float) -> tuple[int, ...]:
return tuple(int(x) for x in str(v).split('.') if x.isdigit())
plugin_version_raw = getattr(sys.modules[namespace], '__archinstall__version__', '0.0')
plugin_version = parse_version(plugin_version_raw)
system_version = parse_version(version_major_and_minor)
def parse_version(v: str | float) -> tuple[int, ...]:
return tuple(int(x) for x in str(v).split('.') if x.isdigit())
if plugin_version < system_version:
error(f'Plugin {sys.modules[namespace]} does not support the current Archinstall version.')
plugin_version = parse_version(plugin_version_raw)
system_version = parse_version(version_major_and_minor)
# Locate the plugin entry-point called Plugin()
# This in accordance with the entry_points() from setup.cfg above
if hasattr(sys.modules[namespace], 'Plugin'):
try:
plugins[namespace] = sys.modules[namespace].Plugin()
info(f'Plugin {plugins[namespace]} has been loaded.')
except Exception as err:
error(
f'Error: {err}',
f'The above error was detected when initiating the plugin: {path}',
)
else:
warn(f"Plugin '{path}' is missing a valid entry-point or is corrupt.")
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()
# This in accordance with the entry_points() from setup.cfg above
if hasattr(sys.modules[namespace], 'Plugin'):
try:
plugins[namespace] = sys.modules[namespace].Plugin()
info(f'Plugin {plugins[namespace]} has been loaded.')
except Exception as err:
error(
f'Error: {err}',
f'The above error was detected when initiating the plugin: {path}',
)
else:
warn(f"Plugin '{path}' is missing a valid entry-point or is corrupt.")

38
tests/test_plugins.py Normal file
View File

@ -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')