diff --git a/archinstall/lib/args.py b/archinstall/lib/args.py index b65024d3..0e788cf5 100644 --- a/archinstall/lib/args.py +++ b/archinstall/lib/args.py @@ -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'): diff --git a/archinstall/lib/plugins.py b/archinstall/lib/plugins.py index 7aa7b6ab..ca77d71c 100644 --- a/archinstall/lib/plugins.py +++ b/archinstall/lib/plugins.py @@ -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.") \ No newline at end of file + 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.") diff --git a/tests/test_plugins.py b/tests/test_plugins.py new file mode 100644 index 00000000..d7c70d1d --- /dev/null +++ b/tests/test_plugins.py @@ -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')