diff --git a/ChangeLog b/ChangeLog index 2419bc91a68..31d66f8b4f2 100644 --- a/ChangeLog +++ b/ChangeLog @@ -1,3 +1,132 @@ +26.2 + - refactor(azure): Extract SSH key cert handling to new certs module + (#6802) [Peyton Robertson] + - fix(alpine): Use Alpine CDN as default APK mirror (#6947) [Natanael Copa] + - fix(oracle): detect iSCSI root via iBFT for dracut images (#6921) + [Gisaldjo Purbollari] (GH: 6915) + - Harden Jinja template rendering with sandboxing (#6888) + [Tyler Coatsworth] + - fix(mounts): escape special characters in fstab mount paths (#6911) + [Eugene Kalinin] (GH: 3603) + - net: add route-metric support in NetworkManager renderer (#6886) + [Harshavardhan Lankipalli] + - test: fix expected empty stderr on cloud-init analyze boot (#6925) + - feat: Fix import error (#6926) [Chloé E.M. Smith] (GH: 6924) + - fix(analyze): warn before GNU date fallback (#6879) + [nanookclaw] (GH: 4357) + - fix(debian): update debian backports suite for bullseye (#6885) + [Challvy Tee] + - test(opennebula): validate gen_conf() output against network-config-v2 + schema (#6855) [Mickaël Canévet] + - ci: add scheduled integration runners for stonking and drop questing + - ci: update PR workflows from questing to latest released ubuntu series + resolute + - ci: bump labeler ver add pull-request write permission for labeling + (#6919) + - fix(tests): make user_groups integration tests distro-aware (#6889) + [Medha Mummigatti] + - chore: bump pylint version, dropping deprecated argparse.FileType + lints (#6917) + - fix(analyze): return integer exit code from analyze_boot (#6863) + [nanookclaw] + - chore: add type annotations to cloudinit.distros.parsers.hosts (#6916) + [ph1n3y] + - test: apt resolute gpg remove accounts for gpg-from-sq (#6902) + - feat(azure): add apply_network_config_set_name option to disable + renames (#6807) [Chris Patterson] + - test: pytest fix strict_parametrization_ids providing required id (#6908) + - docs: fix chpasswd user entries in cloud-config example (#6905) + [Wai Hlyan Min Thein] (GH: 6765) + - fix(distros): clarify misleading ssh_redirect_user warning (#6906) + [Sandhya-d] (GH: 3924) + - chore(ci): address review comments, single operation PRIVATE_KEY creation + - ci: standardize input/dispatch defaults, errors for missing secrets + - ci: rename PYCLOUDLIB_TOML_B64 to indicate base64-encoded + - ci: update image_type defaults, validate required secrets set perms + - ci: switch to pytest-json-ctrf and refactor job env/defaults + - ci: add lxd_vm and EC2 integration test workflows + - ci: require PYCLOUDLIB_TOML and SSH key secrets, write config to + runner temp + - chore(ci): rename and renumber workflow files to 1xx scheme + - feat(opennebula): support global SEARCH_DOMAIN for all interfaces + (#6813) [Mickaël Canévet] + - feat(azure): report failure if missing customdata (#6779) [Cade Jacobson] + - docs: make password lock explicit in users-groups example 3 (#6869) + [Clinton Phillips] + - docs: clarify default user entry in users examples (#6894) + [alan747271363-art] (GH: 6702) + - fix(cmd): enable mypy strict checking for cloudinit.cmd (#6861) + [Borja Velasco Santamaría] + - ai: copilot review comment resolution + - tests: simplify get_syslog_or_console to get_journal_syslog for + systemd-v255 + - test: prefer reading journalctl transport=syslog over /var/log/syslog + - docs: fix invalid badge URLs (#6900) [Richard Coker] (GH: 6899) + - fix(azurelinux): support Azure Linux 4.0 (#6874) [Dan Streetman] + - tests: expect intermittent warning accessing Azure IMDS. Avoid + client.destroy + - fix(cloudstack): get vr information from network manager leases (#6829) + [Ani Sinha] + - fix(cloudstack): get domain name information from network manager + leases (#6829) [Ani Sinha] + - test: fix test_get_domainname_isc_dhclient mock leak (#6829) [Ani Sinha] + - feat(dhcp): Add network manager lease parsing capability (#6829) + [Ani Sinha] + - test: logger specifically configure localhost:514 destination (#6897) + - fix(ug_util): prioritize user-data.users over the default user config + (#6860) [Mostafa Abdelwahab] (GH: 6703) + - ci: fix alpine edge python3.14 tox missing dep py3-python-discovery + (#6864) + - fix(docs): correct users and groups examples 3, 4 and 7 (#6858) + [Bastien Traverse] (GH: 6857) + - feat(opennebula): support ETHx_ROUTES static routes in network config + (#6810) [Mickaël Canévet] + - fix(opennebula): coerce MTU to int in gen_conf() (#6856) + [Mickaël Canévet] + - Enable Amazon Linux to yum_add_repo and ca_certs. (#6767) [joe-usa] + - fix: add single retry of ssh-import-id on exit 1 (#6805) + - fix:(disk_setup):handle empty disk in check_partition_gpt_layout_sfdisk + (#6728) [Amy Chen] (GH: 6682) + - chore(opennebula): enable mypy type checking (#6827) [Mickaël Canévet] + - fix(net): do not resolve IPs when netloc contains a port (#6841) + [Orukaria Ndukiye] + - test: Add new `fake_fs` fixture backed by pyfakefs (#6785) + - feat(bsd): add OpenBSD support to Meson build (#6789) + [Hyacinthe Cartiaux] + - ci: move ssh-import-id tests to github instead of launchpad keys (#6830) + - fix: increase raspberry pi usb timeout (#6823) [Nick Brown] + - fix(debian): Fix locale generation (#6472) [Paul] (GH: 6471) + - fix(ubuntu): Configure arm64 to use archive.ubuntu.com (#6826) + [Dave Jones] (LP: 6825, #2147101) + - test: fix integration test_combined for rhel (#6791) [Amy Chen] + - test: fix test_schema when the schema validation takes longer (#6822) + [Ani Sinha] + - tools: fix brpm script to use correct builddir in temp directory (#6809) + - chore: drop unnecessary pylint ignore settings, use pytest --strict-markers + (#6817) + - ci: disable daily jobs on forks (#6821) + - fix(openbsd): specify the correct usr_lib_exec path (#6793) + [Hyacinthe Cartiaux] + - test: fix missing mocker of rmtree for redhat unittests (#6808) + - chore: drop outdated code (#6798) + - ci: fix integration test CLOUD_INIT_LOCAL_LOG_PATH. drop hashFiles + (#6804) + - chore: auto-format shell (#6780) + - chore(net): log when klibc is parsed (#6799) + - ci: add cloud-init integration test logs upon failure (#6803) + - fix(bsd): correct the _ROOT_TMPDIR path for *BSD system (#6794) + [Hyacinthe Cartiaux] (GH: 5789) + - fix(azure): catch import error as reportable (#6714) + [Cade Jacobson] (GH: 6770) + - WSL: Subprocess cmd.exe with /U to output UTF-16LE (#6717) + [Carlos Nihelton] (GH: 6716) + - feat(util): fail early when hostname is not resolvable in is_resolvable + (#6772) [Alejandro Perez] + - feat(azure): introduce experimental skip_ready_report for Azure (#6771) + [Chris Patterson] + - test: apt tests need to wait_for_cloud_init on potentially slow VMs + - test: use rmadison form devscripts to determine hello pkg version + 26.1 - test: update hello package tested using rmadison (#6774) - test: add details about failing package status to aid in debug (#6775) diff --git a/cloudinit/analyze/dump.py b/cloudinit/analyze/dump.py index 63aab904a39..67ee49cc41f 100644 --- a/cloudinit/analyze/dump.py +++ b/cloudinit/analyze/dump.py @@ -1,12 +1,15 @@ # This file is part of cloud-init. See LICENSE file for license information. import calendar +import logging import sys from datetime import datetime, timezone from typing import IO, Any, Dict, List, Optional, TextIO, Tuple from cloudinit import atomic_helper, subp, util +LOG = logging.getLogger(__name__) + stage_to_description: Dict[str, str] = { "finished": "finished running cloud-init", "init-local": "starting search for local datasources", @@ -76,6 +79,16 @@ def parse_timestamp_from_date(timestampstr: str) -> float: raise ValueError( f"Unable to parse timestamp without GNU date: [{timestampstr}]" ) + # Transitional: shelling out to GNU date is slated for removal in favor of + # native Python parsing. Log the unrecognized timestamp so maintainers can + # collect formats that need direct support (see GH-4357). + LOG.warning( + "analyze: falling back to GNU %s(1) to parse unrecognized " + "timestamp %r; please report this format upstream so it can be " + "handled natively in Python.", + date, + timestampstr, + ) return float( subp.subp([date, "-u", "+%s.%3N", "-d", timestampstr]).stdout.strip() ) diff --git a/cloudinit/config/cc_apk_configure.py b/cloudinit/config/cc_apk_configure.py index 7c4029475ff..d1595e08181 100644 --- a/cloudinit/config/cc_apk_configure.py +++ b/cloudinit/config/cc_apk_configure.py @@ -18,7 +18,7 @@ # If no mirror is specified then use this one -DEFAULT_MIRROR = "https://alpine.global.ssl.fastly.net/alpine" +DEFAULT_MIRROR = "https://dl-cdn.alpinelinux.org/alpine" REPOSITORIES_TEMPLATE = """\ diff --git a/cloudinit/config/cc_mounts.py b/cloudinit/config/cc_mounts.py index 1b4230c51d0..9425fc20c6b 100644 --- a/cloudinit/config/cc_mounts.py +++ b/cloudinit/config/cc_mounts.py @@ -369,7 +369,7 @@ def parse_fstab() -> Tuple[List[str], Dict[str, str], List[str]]: continue toks = line.split() if toks: - fstab_devs[toks[0]] = line + fstab_devs[util.unescape_fstab_field(toks[0])] = line fstab_lines.append(line) return fstab_lines, fstab_devs, fstab_removed @@ -584,7 +584,18 @@ def handle(name: str, cfg: Config, cloud: Cloud, args: list) -> None: LOG.debug("No modifications to fstab needed") return - cfg_lines = ["\t".join(entry) for entry in updated_cfg] + # Only fs_spec (device) and fs_file (mount point) can contain values that + # need escaping; the remaining fstab fields never do, so we skip them. + cfg_lines = [ + "\t".join( + [ + util.escape_fstab_field(entry[0]), + util.escape_fstab_field(entry[1]), + *entry[2:], + ] + ) + for entry in updated_cfg + ] dirs = [d[1] for d in updated_cfg if d[1].startswith("/")] diff --git a/cloudinit/config/schemas/schema-cloud-config-v1.json b/cloudinit/config/schemas/schema-cloud-config-v1.json index 5266adb7d30..7796073c928 100644 --- a/cloudinit/config/schemas/schema-cloud-config-v1.json +++ b/cloudinit/config/schemas/schema-cloud-config-v1.json @@ -1052,8 +1052,8 @@ "properties": { "base_url": { "type": "string", - "default": "https://alpine.global.ssl.fastly.net/alpine", - "description": "The base URL of an Alpine repository, or mirror, to download official packages from. If not specified then it defaults to ``https://alpine.global.ssl.fastly.net/alpine``." + "default": "https://dl-cdn.alpinelinux.org/alpine", + "description": "The base URL of an Alpine repository, or mirror, to download official packages from. If not specified then it defaults to ``https://dl-cdn.alpinelinux.org/alpine``." }, "community_enabled": { "type": "boolean", diff --git a/cloudinit/handlers/jinja_template.py b/cloudinit/handlers/jinja_template.py index b4400304786..23293460719 100644 --- a/cloudinit/handlers/jinja_template.py +++ b/cloudinit/handlers/jinja_template.py @@ -140,7 +140,11 @@ def render_jinja_payload(payload, payload_fn, instance_data, debug=False): ) try: rendered_payload = render_string(payload, instance_jinja_vars) - except (TypeError, exceptions.UndefinedError) as e: + except ( + TypeError, + exceptions.UndefinedError, + exceptions.SecurityError, + ) as e: LOG.warning("Ignoring jinja template for %s: %s", payload_fn, str(e)) return None warnings = [ diff --git a/cloudinit/importer.py b/cloudinit/importer.py index c46df15ce88..d847c7c7ccb 100644 --- a/cloudinit/importer.py +++ b/cloudinit/importer.py @@ -8,7 +8,7 @@ # # This file is part of cloud-init. See LICENSE file for license information. -import importlib +import importlib.util from types import ModuleType from typing import Optional, Sequence @@ -42,10 +42,7 @@ def match_case_insensitive_module_name(mod_name: str) -> str: if not mod_name.startswith("DataSource"): mod_name = f"DataSource{mod_name}" modules = {} - # mypy is right to warn here. This isn't a documented method. - # However it has worked for years without issue, so fixing it is not - # a high priority. We'll fix it when it breaks. - spec = importlib.util.find_spec("cloudinit.sources") # type: ignore + spec = importlib.util.find_spec("cloudinit.sources") if spec and spec.submodule_search_locations: for dir in spec.submodule_search_locations: modules.update(util.get_modules_from_dir(dir)) @@ -70,10 +67,7 @@ def find_module( # Add base name to search paths. Filter out empty paths. full_path = ".".join(filter(None, [path, base_name])) lookup_paths.append(full_path) - # mypy is right to warn here. This isn't a documented method. - # However it has worked for years without issue, so fixing it is not - # a high priority. We'll fix it when it breaks. - if not importlib.util.find_spec(full_path): # type: ignore + if not importlib.util.find_spec(full_path): continue # Check that required_attrs are all present within the module. if _count_attrs(full_path, required_attrs) == len(required_attrs): diff --git a/cloudinit/net/network_manager.py b/cloudinit/net/network_manager.py index d4c705f2726..3ef3cbb80a5 100644 --- a/cloudinit/net/network_manager.py +++ b/cloudinit/net/network_manager.py @@ -434,6 +434,9 @@ def render_interface(self, iface, network_state, renderer): self.config[family]["gateway"] = subnet["gateway"] for route in subnet["routes"]: self._add_route(route) + # metric may apply to both dhcp and static config + if "metric" in subnet: + self.config[family]["route-metric"] = str(subnet["metric"]) # Add subnet-level DNS if "dns_nameservers" in subnet: found_nameservers.extend(subnet["dns_nameservers"]) diff --git a/cloudinit/sources/DataSourceAzure.py b/cloudinit/sources/DataSourceAzure.py index 87b4ab1fbe3..6a9c38a3a95 100644 --- a/cloudinit/sources/DataSourceAzure.py +++ b/cloudinit/sources/DataSourceAzure.py @@ -19,7 +19,7 @@ import requests -from cloudinit import net, performance, sources, ssh_util, subp, util +from cloudinit import net, performance, sources, subp, util from cloudinit.config import cc_mounts from cloudinit.event import EventScope, EventType from cloudinit.net import device_driver @@ -30,7 +30,7 @@ ) from cloudinit.net.ephemeral import EphemeralDHCPv4, EphemeralIPv4Network from cloudinit.reporting import events -from cloudinit.sources.azure import errors, identity, imds, kvp +from cloudinit.sources.azure import certs, errors, identity, imds, kvp from cloudinit.sources.helpers import netlink from cloudinit.sources.helpers.azure import ( DEFAULT_WIRESERVER_ENDPOINT, @@ -1077,7 +1077,9 @@ def _get_public_keys_from_imds(self, imds_md: dict) -> List[str]: report_diagnostic_event(log_msg, logger_func=LOG.debug) raise - if any(not _key_is_openssh_formatted(key=key) for key in ssh_keys): + ssh_keys = [certs.sanitize_openssh_key(key) for key in ssh_keys] + + if any(not certs.is_openssh_formatted(key) for key in ssh_keys): log_msg = "Key(s) not in OpenSSH format" report_diagnostic_event(log_msg, logger_func=LOG.debug) raise ValueError(log_msg) @@ -1784,23 +1786,6 @@ def _disable_password_from_imds(imds_data): return None -def _key_is_openssh_formatted(key): - """ - Validate whether or not the key is OpenSSH-formatted. - """ - # See https://bugs.launchpad.net/cloud-init/+bug/1910835 - if "\r\n" in key.strip(): - return False - - parser = ssh_util.AuthKeyLineParser() - try: - akl = parser.parse(key) - except TypeError: - return False - - return akl.keytype is not None - - def _partitions_on_device(devpath, maxnum=16): # return a list of tuples (ptnum, path) for each part on devpath for suff in ("-part", "p", ""): diff --git a/cloudinit/sources/DataSourceOracle.py b/cloudinit/sources/DataSourceOracle.py index 4a846a897b8..e19b30ea21f 100644 --- a/cloudinit/sources/DataSourceOracle.py +++ b/cloudinit/sources/DataSourceOracle.py @@ -14,6 +14,7 @@ """ import base64 +import glob import ipaddress import json import logging @@ -48,6 +49,12 @@ # indicates that an MTU of 9000 is used within OCI MTU = 9000 +# iBFT target flags exposed by the kernel's iscsi_ibft module. A target that +# is both valid and firmware-boot-selected indicates an iSCSI boot device. +IBFT_TARGET_FLAGS_GLOB = "/sys/firmware/ibft/target*/flags" +IBFT_TGT_BLOCK_VALID = 0x01 +IBFT_TGT_FIRMWARE_BOOT_SELECTED = 0x02 + class ReadOpcMetadataResponse(NamedTuple): version: int @@ -68,6 +75,24 @@ def _is_applicable(self) -> bool: return bool(self._files) +def _ibft_has_iscsi_boot_target() -> bool: + """Return True if an iBFT target is flagged as a firmware boot device.""" + for flags_path in glob.glob(IBFT_TARGET_FLAGS_GLOB): + try: + flags = int(util.load_text_file(flags_path).strip()) + if ( + flags & IBFT_TGT_BLOCK_VALID + and flags & IBFT_TGT_FIRMWARE_BOOT_SELECTED + ): + LOG.debug( + "Detected iSCSI boot target via iBFT: %s", flags_path + ) + return True + except (OSError, ValueError): + continue + return False + + def _ensure_netfailover_safe(network_config: NetworkConfig) -> None: """ Search network config physical interfaces to see if any of them are @@ -216,7 +241,9 @@ def _get_data(self): ) else: network_context = util.nullcontext() - fetch_primary_nic = not self._is_iscsi_root() + # Use the klibc source directly rather than _is_iscsi_root: iBFT may + # report iSCSI root even when no klibc files exist to render config. + fetch_primary_nic = not self._network_config_source.is_applicable() fetch_secondary_nics = self.ds_cfg.get( "configure_secondary_nics", BUILTIN_DS_CONFIG["configure_secondary_nics"], @@ -274,7 +301,7 @@ def get_public_ssh_keys(self): def _is_iscsi_root(self) -> bool: """Return whether we are on a iscsi machine.""" - return self._network_config_source.is_applicable() + return _ibft_has_iscsi_boot_target() def _get_iscsi_config(self) -> dict: return self._network_config_source.render_config() @@ -293,15 +320,8 @@ def network_config(self): return self._network_config set_primary = False - if self._is_iscsi_root(): + if self._network_config_source.is_applicable(): self._network_config = self._get_iscsi_config() - logging.debug( - "Instance is using iSCSI root, setting primary NIC as critical" - ) - # This is necessary for Oracle baremetal instances in case they are - # running on an IPv6-only network. Without this, they become - # unreachable/unrecoverable after a shutdown. - self._network_config["config"][0]["keep_configuration"] = True if not self._has_network_config(): LOG.debug( "Could not obtain network configuration from initramfs. " @@ -324,6 +344,14 @@ def network_config(self): "Failed to parse IMDS network configuration!", ) + # On iSCSI root, mark the primary NIC as critical so it is not torn + # down on shutdown, whether config came from initramfs or IMDS. + if self._is_iscsi_root() and self._has_network_config(): + LOG.debug( + "Instance is using iSCSI root, setting primary NIC as critical" + ) + self._network_config["config"][0]["keep_configuration"] = True + # we need to verify that the nic selected is not a netfail over # device and, if it is a netfail master, then we need to avoid # emitting any match by mac diff --git a/cloudinit/sources/azure/certs.py b/cloudinit/sources/azure/certs.py new file mode 100644 index 00000000000..74ebb34ff59 --- /dev/null +++ b/cloudinit/sources/azure/certs.py @@ -0,0 +1,145 @@ +# Copyright (C) 2024 Microsoft Corporation. +# +# This file is part of cloud-init. See LICENSE file for license information. + +import logging +import re +from typing import List, Optional + +from cloudinit import ssh_util, subp + +LOG = logging.getLogger(__name__) + +_CERTIFICATE_BLOCK_RE = re.compile( + r"-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----", + re.DOTALL, +) + + +def sanitize_openssh_key(key: str) -> str: + r"""Sanitize an OpenSSH key by removing embedded CRLF sequences. + + Azure-generated SSH keys may contain CRLF (\r\n) sequences embedded + in the base64 key data. This strips those sequences so the key can + be properly parsed and written to authorized_keys. + + Note: Azure no longer generates keys with embedded CRLF sequences, + but previously provisioned keys may still be in use. + + See https://bugs.launchpad.net/cloud-init/+bug/1910835 + """ + key = key.strip() + if "\r\n" in key: + LOG.debug("SSH key contains embedded CRLF sequences, sanitizing.") + key = key.replace("\r\n", "") + return key + + +def is_openssh_formatted(key: str) -> bool: + """Validate whether or not the key is OpenSSH-formatted. + + This checks if a given string is a valid OpenSSH public key format + (e.g., ssh-rsa, ssh-ed25519, ecdsa-sha2-nistp256). + + """ + if not key: + LOG.debug("Empty SSH key content provided.") + return False + + parser = ssh_util.AuthKeyLineParser() + try: + akl = parser.parse(key) + except TypeError: + LOG.debug("SSH key could not be parsed.") + return False + + if akl.keytype is None: + LOG.debug("SSH key type is missing.") + return False + + return True + + +def is_x509_certificate(cert: str) -> bool: + """Check if the input string is an x509 certificate in PEM format. + + This validates that the certificate is a valid x509 certificate by + attempting to parse it with openssl. + """ + if not cert: + LOG.debug("Empty certificate provided.") + return False + + cert = cert.strip() + + if "-----BEGIN CERTIFICATE-----" not in cert: + LOG.debug("No BEGIN CERTIFICATE marker.") + return False + + if "-----END CERTIFICATE-----" not in cert: + LOG.debug("No END CERTIFICATE marker.") + return False + + # Attempt to parse the certificate with openssl to validate it. + try: + cmd = ["openssl", "x509", "-noout", "-text"] + subp.subp(cmd, data=cert) + return True + except subp.ProcessExecutionError as e: + LOG.debug("Certificate could not be parsed: %s", e) + return False + + +def extract_x509_certificates(data: Optional[str]) -> List[str]: + """Extract and validate all x509 certificates from a data bundle. + + The data may contain a mix of certificates and private keys. This function + finds all valid x509 certificates and returns them. + + Args: + data: String containing certificate data, potentially mixed with + private keys or other content. May be None. + + Returns: + A list of valid x509 certificate strings. Empty if none are found. + """ + if not data: + LOG.debug("No data provided for certificate extraction.") + return [] + + certificates = [] + for match in _CERTIFICATE_BLOCK_RE.finditer(data): + certificate = match.group(0) + if is_x509_certificate(certificate): + LOG.debug("Successfully extracted x509 certificate from bundle.") + certificates.append(certificate) + else: + LOG.debug( + "Found certificate block but validation failed, skipping." + ) + + if not certificates: + LOG.debug("No valid x509 certificates found in data bundle.") + + return certificates + + +def convert_x509_to_openssh(certificate: str) -> str: + """Convert an x509 certificate to OpenSSH public key format.""" + LOG.debug("Converting x509 certificate to OpenSSH public key format.") + openssl_cmd = ["openssl", "x509", "-noout", "-pubkey"] + try: + pub_key, _ = subp.subp(openssl_cmd, data=certificate) + except subp.ProcessExecutionError as e: + LOG.warning("Failed to extract public key from certificate: %s", e) + raise + + keygen_cmd = ["ssh-keygen", "-i", "-m", "PKCS8", "-f", "/dev/stdin"] + try: + ssh_key, _ = subp.subp(keygen_cmd, data=pub_key) + except subp.ProcessExecutionError as e: + LOG.warning("Failed to convert public key to OpenSSH format: %s", e) + raise + + LOG.debug("Successfully converted x509 certificate to OpenSSH format.") + return ssh_key diff --git a/cloudinit/sources/helpers/azure.py b/cloudinit/sources/helpers/azure.py index 84f8f10e57d..0344f60a2ee 100644 --- a/cloudinit/sources/helpers/azure.py +++ b/cloudinit/sources/helpers/azure.py @@ -6,7 +6,6 @@ import json import logging import os -import re import textwrap import zlib from contextlib import contextmanager @@ -18,7 +17,7 @@ from cloudinit import distros, subp, temp_utils, url_helper, util, version from cloudinit.reporting import events -from cloudinit.sources.azure import errors +from cloudinit.sources.azure import certs, errors LOG = logging.getLogger(__name__) @@ -490,10 +489,7 @@ def _run_x509_action(action, cert): @azure_ds_telemetry_reporter def _get_ssh_key_from_cert(self, certificate): - pub_key = self._run_x509_action("-pubkey", certificate) - keygen_cmd = ["ssh-keygen", "-i", "-m", "PKCS8", "-f", "/dev/stdin"] - ssh_key, _ = subp.subp(keygen_cmd, data=pub_key) - return ssh_key + return certs.convert_x509_to_openssh(certificate) @azure_ds_telemetry_reporter def _get_fingerprint_from_cert(self, certificate): @@ -539,19 +535,13 @@ def parse_certificates(self, certificates_xml): """Given the Certificates XML document, return a dictionary of fingerprints and associated SSH keys derived from the certs.""" out = self._decrypt_certs_from_xml(certificates_xml) - current = [] keys = {} - for line in out.splitlines(): - current.append(line) - if re.match(r"[-]+END .*?KEY[-]+$", line): - # ignore private_keys - current = [] - elif re.match(r"[-]+END .*?CERTIFICATE[-]+$", line): - certificate = "\n".join(current) - ssh_key = self._get_ssh_key_from_cert(certificate) - fingerprint = self._get_fingerprint_from_cert(certificate) - keys[fingerprint] = ssh_key - current = [] + + for certificate in certs.extract_x509_certificates(out): + ssh_key = self._get_ssh_key_from_cert(certificate) + fingerprint = self._get_fingerprint_from_cert(certificate) + keys[fingerprint] = ssh_key + return keys diff --git a/cloudinit/templater.py b/cloudinit/templater.py index d9853932bc3..7d90e45901b 100644 --- a/cloudinit/templater.py +++ b/cloudinit/templater.py @@ -17,7 +17,8 @@ import re import sys -from jinja2 import DebugUndefined, Template, TemplateSyntaxError +from jinja2 import DebugUndefined, TemplateSyntaxError +from jinja2.sandbox import SandboxedEnvironment from cloudinit import performance from cloudinit import type_utils as tu @@ -134,15 +135,12 @@ def jinja_render(content, params): add = "\n" if content.endswith("\n") else "" try: with performance.Timed("Rendering jinja2 template"): - return ( - Template( - content, - undefined=UndefinedJinjaVariable, - trim_blocks=True, - extensions=["jinja2.ext.do"], - ).render(**params) - + add + jinja_env = SandboxedEnvironment( + undefined=UndefinedJinjaVariable, + trim_blocks=True, + extensions=["jinja2.ext.do"], ) + return jinja_env.from_string(content).render(**params) + add except TemplateSyntaxError as template_syntax_error: template_syntax_error.lineno += 1 raise JinjaSyntaxParsingException( diff --git a/cloudinit/util.py b/cloudinit/util.py index 2cdd73655ef..5fa0f94c73a 100644 --- a/cloudinit/util.py +++ b/cloudinit/util.py @@ -1910,6 +1910,35 @@ def unmounter(umount): subp.subp(umount_cmd) +# Per fstab(5), fstab fields are whitespace-separated, so these characters +# must be octal-escaped when they appear inside a field. Backslash must come +# first so we don't double-escape the escapes we introduce. +_FSTAB_ESCAPES = ( + ("\\", "\\134"), + (" ", "\\040"), + ("\t", "\\011"), + ("\n", "\\012"), +) + + +def escape_fstab_field(value: str) -> str: + """Octal-escape special characters for safe writing to fstab.""" + for char, escaped in _FSTAB_ESCAPES: + value = value.replace(char, escaped) + return value + + +def unescape_fstab_field(value: str) -> str: + """Reverse escape_fstab_field. + + Iterate _FSTAB_ESCAPES in reverse so backslash is decoded last; this + avoids mis-decoding a field whose original value contained a backslash. + """ + for char, escaped in reversed(_FSTAB_ESCAPES): + value = value.replace(escaped, char) + return value + + def mounts(): mounted = {} try: @@ -1938,9 +1967,9 @@ def mounts(): mp = m.group(2) fstype = m.group(3) opts = m.group(4) - # If the name of the mount point contains spaces these - # can be escaped as '\040', so undo that.. - mp = mp.replace("\\040", " ") + # Mount points may contain octal-escaped characters (e.g. a + # space as '\040'); undo that so callers see the real path. + mp = unescape_fstab_field(mp) mounted[dev] = { "fstype": fstype, "mountpoint": mp, diff --git a/debian/changelog b/debian/changelog index c5d5881f4c4..2ab52523062 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,4 +1,4 @@ -cloud-init (26.1-0ubuntu1~24.04.2) UNRELEASED; urgency=medium +cloud-init (26.2-0ubuntu1~24.04.1) noble; urgency=medium * d/control: add python3-pyfakefs dependency * refresh patches: @@ -7,8 +7,13 @@ cloud-init (26.1-0ubuntu1~24.04.2) UNRELEASED; urgency=medium - d/p/no-single-process.patch - d/p/retain-setuptools.patch. Use PACKAGED_VERSION environment variable. * Upstream snapshot based on upstream/main at 399cc042. + * refresh patches: + - d/p/no-nocloud-network.patch + * Upstream snapshot based on 26.2. (LP: #2164685). + List of changes from upstream can be found at + https://raw.githubusercontent.com/canonical/cloud-init/26.2/ChangeLog - -- Chad Smith Mon, 06 Jul 2026 16:41:25 -0600 + -- Chad Smith Thu, 20 Aug 2026 15:02:15 -0600 cloud-init (26.1-0ubuntu1~24.04.1) noble; urgency=medium diff --git a/debian/patches/no-nocloud-network.patch b/debian/patches/no-nocloud-network.patch index b687d45fbd6..14418ede112 100644 --- a/debian/patches/no-nocloud-network.patch +++ b/debian/patches/no-nocloud-network.patch @@ -54,7 +54,7 @@ Last-Update: 2024-08-02 ) --- a/tests/unittests/test_util.py +++ b/tests/unittests/test_util.py -@@ -2448,7 +2448,7 @@ class TestReadOptionalSeed: +@@ -2486,7 +2486,7 @@ class TestReadOptionalSeed: { "meta-data": {"md": "val"}, "user-data": b"ud", @@ -63,7 +63,7 @@ Last-Update: 2024-08-02 "vendor-data": None, }, True, -@@ -2503,7 +2503,7 @@ class TestReadSeeded: +@@ -2541,7 +2541,7 @@ class TestReadSeeded: assert found_md == {"key1": "val1"} assert found_ud == ud assert found_vd == vd @@ -72,7 +72,7 @@ Last-Update: 2024-08-02 @pytest.mark.parametrize( "base, feature_flag, req_urls", -@@ -2512,7 +2512,6 @@ class TestReadSeeded: +@@ -2550,7 +2550,6 @@ class TestReadSeeded: "http://10.0.0.1/%s?qs=1", True, [ @@ -80,7 +80,7 @@ Last-Update: 2024-08-02 "http://10.0.0.1/meta-data?qs=1", "http://10.0.0.1/user-data?qs=1", "http://10.0.0.1/vendor-data?qs=1", -@@ -2523,7 +2522,6 @@ class TestReadSeeded: +@@ -2561,7 +2560,6 @@ class TestReadSeeded: "https://10.0.0.1:8008/", True, [ @@ -88,7 +88,7 @@ Last-Update: 2024-08-02 "https://10.0.0.1:8008/meta-data", "https://10.0.0.1:8008/user-data", "https://10.0.0.1:8008/vendor-data", -@@ -2534,7 +2532,6 @@ class TestReadSeeded: +@@ -2572,7 +2570,6 @@ class TestReadSeeded: "https://10.0.0.1:8008", True, [ @@ -96,7 +96,7 @@ Last-Update: 2024-08-02 "https://10.0.0.1:8008/meta-data", "https://10.0.0.1:8008/user-data", "https://10.0.0.1:8008/vendor-data", -@@ -2545,7 +2542,6 @@ class TestReadSeeded: +@@ -2583,7 +2580,6 @@ class TestReadSeeded: "https://10.0.0.1:8008", False, [ @@ -104,7 +104,7 @@ Last-Update: 2024-08-02 "https://10.0.0.1:8008meta-data", "https://10.0.0.1:8008user-data", "https://10.0.0.1:8008vendor-data", -@@ -2556,7 +2552,6 @@ class TestReadSeeded: +@@ -2594,7 +2590,6 @@ class TestReadSeeded: "https://10.0.0.1:8008?qs=", True, [ @@ -112,7 +112,7 @@ Last-Update: 2024-08-02 "https://10.0.0.1:8008?qs=meta-data", "https://10.0.0.1:8008?qs=user-data", "https://10.0.0.1:8008?qs=vendor-data", -@@ -2595,7 +2590,7 @@ class TestReadSeeded: +@@ -2633,7 +2628,7 @@ class TestReadSeeded: # user-data, vendor-data read raw. It could be scripts or other format assert found_ud == "/user-data: 1" assert found_vd == "/vendor-data: 1" @@ -121,7 +121,7 @@ Last-Update: 2024-08-02 assert [ mock.call(req_url, timeout=5, retries=10) for req_url in req_urls ] == m_read.call_args_list -@@ -2620,7 +2615,7 @@ class TestReadSeededWithoutVendorData: +@@ -2658,7 +2653,7 @@ class TestReadSeededWithoutVendorData: assert found_md == {"key1": "val1"} assert found_ud == ud assert found_vd == vd diff --git a/doc/module-docs/cc_mounts/data.yaml b/doc/module-docs/cc_mounts/data.yaml index 7548bddaece..059dbbb9239 100644 --- a/doc/module-docs/cc_mounts/data.yaml +++ b/doc/module-docs/cc_mounts/data.yaml @@ -11,7 +11,12 @@ cc_mounts: omitted. Any mounts that do not appear to either an attached block device or network resource will be skipped with a log like "Ignoring nonexistent mount ...". - + + Whitespace and other special characters in the ``fs_spec`` and ``fs_file`` + fields are automatically octal-escaped when written to ``/etc/fstab`` (for + example, a space becomes ``\040``), so mount points containing spaces work + without any manual escaping in the config. + Cloud-init will attempt to add the following mount directives if available and unconfigured in ``/etc/fstab``: @@ -65,5 +70,11 @@ cc_mounts: Example 2: Create a 2 GB swap file at ``/swapfile`` using human-readable values. file: cc_mounts/example2.yaml + - comment: > + Example 3: Mount a device at a mount point whose path contains a space. + Whitespace and other special characters in the ``fs_spec`` and + ``fs_file`` fields are automatically escaped (e.g. a space is written as + ``\040``) so that the resulting ``/etc/fstab`` entry can be parsed. + file: cc_mounts/example3.yaml name: Mounts title: Configure mount points and swap files diff --git a/doc/module-docs/cc_mounts/example3.yaml b/doc/module-docs/cc_mounts/example3.yaml new file mode 100644 index 00000000000..a08821a6ce3 --- /dev/null +++ b/doc/module-docs/cc_mounts/example3.yaml @@ -0,0 +1,3 @@ +#cloud-config +mounts: +- [ /dev/sr0, "/mnt/Cdrom Drive", auto, "uid=1000,gid=1000", "0", "0" ] diff --git a/meson.build b/meson.build index 0e06d8de292..27c3abe09c8 100644 --- a/meson.build +++ b/meson.build @@ -1,6 +1,6 @@ project( 'cloud-init', - version: '26.1', + version: '26.2', meson_version: '>=0.63.0', # rockylinux/9 license: 'GPL-3 OR Apache-2.0', default_options: [ diff --git a/templates/sources.list.debian.tmpl b/templates/sources.list.debian.tmpl index e8a7c01501d..3c03af5c3e8 100644 --- a/templates/sources.list.debian.tmpl +++ b/templates/sources.list.debian.tmpl @@ -26,5 +26,8 @@ deb-src {{mirror}} {{codename}}-updates main ## N.B. software from this repository may not have been tested as ## extensively as that contained in the main release, although it includes ## newer versions of some applications which may provide useful features. +## N.B. backports repository was deprecated in Debian 11 (bullseye). +{% if codename != 'bullseye' -%} deb {{mirror}} {{codename}}-backports main deb-src {{mirror}} {{codename}}-backports main +{% endif -%} diff --git a/tests/integration_tests/bugs/test_gh3603.py b/tests/integration_tests/bugs/test_gh3603.py new file mode 100644 index 00000000000..eca7ff0b844 --- /dev/null +++ b/tests/integration_tests/bugs/test_gh3603.py @@ -0,0 +1,45 @@ +"""Integration test for gh-3603. + +cc_mounts historically wrote /etc/fstab entries without escaping, so a +space in the device (fs_spec) or mount point (fs_file) produced an +unparseable line and broke ``mount -a``. Verify that a mount point +containing a space is octal-escaped in /etc/fstab, that cloud-init logs +stay clean, and that the real (unescaped) directory is created. + +A network fs_spec (contains ``:``) with ``noauto`` keeps the entry in +fstab without cloud-init attempting the actual mount, so the test does +not depend on any real device or NFS server. + +https://github.com/canonical/cloud-init/issues/3603 +""" + +import pytest + +from tests.integration_tests.instances import IntegrationInstance +from tests.integration_tests.util import verify_clean_log + +USER_DATA = """\ +#cloud-config +mounts: +- - server.example:/export + - /mnt/Cdrom Drive + - nfs + - defaults,noauto + - "0" + - "0" +""" + + +@pytest.mark.user_data(USER_DATA) +def test_mount_point_with_space_is_escaped(client: IntegrationInstance): + # cloud-init should have configured mounts without any errors. + log = client.read_from_file("/var/log/cloud-init.log") + verify_clean_log(log) + assert "SUCCESS: config-mounts ran successfully" in log + + # The space must be octal-escaped (\040) in the /etc/fstab entry. + fstab = client.read_from_file("/etc/fstab") + assert "/mnt/Cdrom\\040Drive" in fstab + + # The real directory (with a literal space) must have been created. + assert client.execute("test -d '/mnt/Cdrom Drive'").ok diff --git a/tests/integration_tests/cmd/test_analyze.py b/tests/integration_tests/cmd/test_analyze.py index ceee47e425a..638ffce912d 100644 --- a/tests/integration_tests/cmd/test_analyze.py +++ b/tests/integration_tests/cmd/test_analyze.py @@ -21,7 +21,10 @@ def test_analyze_boot_ordered_timestamps(self, module_client): """ assert module_client.execute("cloud-init status --wait --long").ok result = module_client.execute("cloud-init analyze boot") - assert result.stderr == "container" + assert ( + result.ok + ), f"cloud-init analyze boot unexpected exit [{result.return_code}]" + assert result.stderr == "" container_start_time = get_datetime_from_string( result.stdout, "^\\s*Container started at: (.+?)$" diff --git a/tests/unittests/analyze/test_dump.py b/tests/unittests/analyze/test_dump.py index 45bf1c743e4..b5d1adb8799 100644 --- a/tests/unittests/analyze/test_dump.py +++ b/tests/unittests/analyze/test_dump.py @@ -1,5 +1,6 @@ # This file is part of cloud-init. See LICENSE file for license information. +import logging import warnings from contextlib import suppress from datetime import datetime, timezone @@ -13,7 +14,9 @@ has_gnu_date, parse_ci_logline, parse_timestamp, + parse_timestamp_from_date, ) +from cloudinit.subp import SubpResult from cloudinit.util import write_file @@ -112,6 +115,61 @@ def test_parse_timestamp_handles_explicit_timezone(self): == "2020-09-12 12:39:20" ) + @mock.patch("cloudinit.analyze.dump.subp.subp") + @mock.patch("cloudinit.analyze.dump.has_gnu_date", return_value=True) + @mock.patch("cloudinit.analyze.dump.util.is_Linux", return_value=True) + def test_parse_timestamp_from_date_warns_on_gnu_date_fallback( + self, _m_is_linux, _m_has_gnu, m_subp, caplog + ): + """A warning is emitted whenever GNU date(1) is used as a fallback.""" + m_subp.return_value = SubpResult("1597333950.000\n", "") + timestampstr = "17:15 08/13" + with caplog.at_level(logging.WARNING, logger="cloudinit.analyze.dump"): + assert 1597333950.0 == parse_timestamp_from_date(timestampstr) + m_subp.assert_called_once_with( + ["date", "-u", "+%s.%3N", "-d", timestampstr] + ) + warnings_logged = [ + r for r in caplog.records if r.levelname == "WARNING" + ] + assert 1 == len(warnings_logged) + assert "date(1)" in warnings_logged[0].getMessage() + assert repr(timestampstr) in warnings_logged[0].getMessage() + + @mock.patch("cloudinit.analyze.dump.subp.which", return_value="/bin/gdate") + @mock.patch("cloudinit.analyze.dump.subp.subp") + @mock.patch("cloudinit.analyze.dump.has_gnu_date", return_value=False) + @mock.patch("cloudinit.analyze.dump.util.is_Linux", return_value=False) + def test_parse_timestamp_from_date_warns_on_gdate_fallback( + self, _m_is_linux, _m_has_gnu, m_subp, _m_which, caplog + ): + """The warning identifies gdate when used on non-Linux systems.""" + m_subp.return_value = SubpResult("1597333950.000\n", "") + timestampstr = "17:15 08/13" + with caplog.at_level(logging.WARNING, logger="cloudinit.analyze.dump"): + parse_timestamp_from_date(timestampstr) + m_subp.assert_called_once_with( + ["gdate", "-u", "+%s.%3N", "-d", timestampstr] + ) + warnings_logged = [ + r for r in caplog.records if r.levelname == "WARNING" + ] + assert 1 == len(warnings_logged) + assert "gdate(1)" in warnings_logged[0].getMessage() + assert repr(timestampstr) in warnings_logged[0].getMessage() + + @mock.patch("cloudinit.analyze.dump.subp.which", return_value=None) + @mock.patch("cloudinit.analyze.dump.has_gnu_date", return_value=False) + @mock.patch("cloudinit.analyze.dump.util.is_Linux", return_value=True) + def test_parse_timestamp_from_date_no_warning_when_no_gnu_date( + self, _m_is_linux, _m_has_gnu, _m_which, caplog + ): + """No warning is logged when there is no GNU date to fall back to.""" + with caplog.at_level(logging.WARNING, logger="cloudinit.analyze.dump"): + with pytest.raises(ValueError): + parse_timestamp_from_date("17:15 08/13") + assert [] == [r for r in caplog.records if r.levelname == "WARNING"] + class TestParseCILogLine: def test_parse_logline_returns_none_without_separators(self): diff --git a/tests/unittests/config/test_cc_apk_configure.py b/tests/unittests/config/test_cc_apk_configure.py index db78687e3d3..5aa045cf5e4 100644 --- a/tests/unittests/config/test_cc_apk_configure.py +++ b/tests/unittests/config/test_cc_apk_configure.py @@ -20,7 +20,7 @@ from tests.unittests.util import get_cloud REPO_FILE = "/etc/apk/repositories" -DEFAULT_MIRROR_URL = "https://alpine.global.ssl.fastly.net/alpine" +DEFAULT_MIRROR_URL = "https://dl-cdn.alpinelinux.org/alpine" CC_APK = "cloudinit.config.cc_apk_configure" diff --git a/tests/unittests/config/test_cc_mounts.py b/tests/unittests/config/test_cc_mounts.py index 80dbf5c6e11..8333ffe3e0b 100644 --- a/tests/unittests/config/test_cc_mounts.py +++ b/tests/unittests/config/test_cc_mounts.py @@ -535,6 +535,52 @@ def test_fstab_mounts_combinations(self): ).strip() ) + def test_fstab_mountpoint_with_spaces(self): + """Spaces in the fs_spec and fs_file fields are octal-escaped. + + Reproduces GH-3603: an unescaped space in the device or mount point + breaks `mount -a` parsing of /etc/fstab. + """ + cfg = { + "mounts": [ + [ + "/dev/sr0 spaced", + "/mnt/Cdrom Drive", + "auto", + "uid=1000,gid=1000", + "0", + "0", + ], + ] + } + cc_mounts.handle("", cfg, self.mock_cloud, []) + with open(cc_mounts.FSTAB_PATH, "r") as fd: + fstab_new_content = fd.read() + assert ( + "/dev/sr0\\040spaced\t/mnt/Cdrom\\040Drive\tauto\t" + "uid=1000,gid=1000,comment=cloudconfig\t0\t0\n" + ) in fstab_new_content + + def test_fstab_mountpoint_with_spaces_idempotent(self): + """A second run must not duplicate an escaped mount entry.""" + cfg = { + "mounts": [ + [ + "/dev/sr0", + "/mnt/Cdrom Drive", + "auto", + "uid=1000,gid=1000", + "0", + "0", + ], + ] + } + cc_mounts.handle("", cfg, self.mock_cloud, []) + cc_mounts.handle("", cfg, self.mock_cloud, []) + with open(cc_mounts.FSTAB_PATH, "r") as fd: + fstab_new_content = fd.read() + assert fstab_new_content.count("/mnt/Cdrom\\040Drive") == 1 + class TestCreateSwapfile: @pytest.mark.parametrize("fstype", ("xfs", "btrfs", "ext4", "other")) diff --git a/tests/unittests/net/test_network_manager.py b/tests/unittests/net/test_network_manager.py index 694c468e5fd..301582454ef 100644 --- a/tests/unittests/net/test_network_manager.py +++ b/tests/unittests/net/test_network_manager.py @@ -22,6 +22,56 @@ def _parse_network_state_from_config(self, config): config = yaml.safe_load(config) return network_state.parse_net_config_data(config) + def test_subnet_route_metric(self, tmpdir): + """Test subnet metric renders as route-metric in NM config.""" + config = textwrap.dedent( + """\ + version: 1 + config: + - mac_address: 'xx:xx:xx:xx:xx:00' + name: ens1f0np0 + type: physical + subnets: + - type: static + address: 192.168.1.10/24 + metric: 100 + """ + ) + + expected_config = { + "/etc/NetworkManager/system-connections/cloud-init-ens1f0np0.nmconnection": textwrap.dedent( # noqa: E501 + """\ + # Generated by cloud-init. Changes will be lost. + + [connection] + id=cloud-init ens1f0np0 + uuid=99c4bf6c-1691-53c4-bfe8-abdcb90b278a + autoconnect-priority=120 + type=ethernet + + [user] + org.freedesktop.NetworkManager.origin=cloud-init + + [ethernet] + mac-address=XX:XX:XX:XX:XX:00 + + [ipv4] + method=manual + may-fail=false + address1=192.168.1.10/24 + route-metric=100 + + """ + ), + } + + with mock.patch("cloudinit.net.get_interfaces_by_mac"): + ns = self._parse_network_state_from_config(config) + target = str(tmpdir) + network_manager.Renderer().render_network_state(ns, target=target) + rendered_content = dir2dict(target) + assert_equal_dict(expected_config, rendered_content) + def test_bond_dns_baseline(self, tmpdir): config = textwrap.dedent( diff --git a/tests/unittests/sources/azure/test_certs.py b/tests/unittests/sources/azure/test_certs.py new file mode 100644 index 00000000000..b3d09991a30 --- /dev/null +++ b/tests/unittests/sources/azure/test_certs.py @@ -0,0 +1,394 @@ +# This file is part of cloud-init. See LICENSE file for license information. + +import shutil +from pathlib import Path +from textwrap import dedent +from unittest import mock + +import pytest + +from cloudinit import subp +from cloudinit.sources.azure import certs + + +@pytest.fixture +def data_file_path(): + """Return a helper that resolves Azure test data file paths.""" + + def _path(name): + return Path("tests", "data", "azure", name) + + return _path + + +@pytest.fixture +def cert_data(data_file_path): + """Load the test certificate data, skipping if unavailable.""" + cert_file = data_file_path("pubkey_extract_cert") + if not cert_file.exists(): + pytest.skip("Test data file not found") + return cert_file.read_text() + + +# --- Shared test data --- + +_VALID_RSA_KEY = ( + "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDHU9IDclbKVYVbYuv0+zViX" + "+wTwlKspslmy/uf3hkWLh7pyzyrq70S7qtSW2EGixUPxZS/R8pOLHoinlKF9ILgj" + "0gVTCJsSwnWpXRg3rhZwIVoYMHN50BHS1SqVD0lsWNMXmo76LoJcjmWvwIznvj5C" + "/gnhU+K7+c3m7AlCyU2wjwpBAEYj7PQs6l/wTqpEiaqC5NytNBd7qp+lYYysVrpa" + "1PFL0Nj4MMZARIfjkiJtL9qDhy9YZeJRQ6q/Fhz0kjvkZnfxixfKF4yWzOfhBrAt" + "pF6oOnuYKk3hxjh9KjTTX4/U8zdLojalX09iyHyEjwJKGlGEpzh1aY7t5btUyvp" +) + +_VALID_ED25519_KEY = ( + "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIOMqqnkVzrm0SdG6U" + "Orhxd+wTwlKspslmy/uf user@host" +) + +_VALID_ECDSA_KEY = ( + "ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAI" + "bmlzdHAyNTYAAABBBEmKSENjQEezOmxkZMy7opKgwFB9nkt5YRrYMjNuG5N87u" + "RFNngSmOjp2S185mF user@host" +) + +_INVALID_X509_CERT = ( + "-----BEGIN CERTIFICATE-----\nINVALID\n-----END CERTIFICATE-----" +) + +_X509_CERT = dedent( + """\ + -----BEGIN CERTIFICATE----- + MIIB+TCCAeOgAwIBAgIBATANBgkqhkiG9w0BAQUFADAWMRQwEgYDVQQDDAtSb290 + -----END CERTIFICATE----- + """ +) + + +class TestIsOpensshFormatted: + """Test is_openssh_formatted() function.""" + + @pytest.mark.parametrize( + "key", + [ + pytest.param(_VALID_RSA_KEY, id="rsa-without-comment"), + pytest.param(_VALID_ED25519_KEY, id="ed25519"), + pytest.param(_VALID_ECDSA_KEY, id="ecdsa"), + pytest.param( + "ssh-rsa AAAAB3NzaC1yc2EAAAA\r\nBBBB user@host", + id="windows-line-endings", + ), + ], + ) + def test_valid(self, key): + """Valid OpenSSH keys should return True.""" + assert certs.is_openssh_formatted(key) is True + + @pytest.mark.parametrize( + "value", + [ + pytest.param(_X509_CERT, id="x509-certificate"), + pytest.param("", id="empty-string"), + pytest.param("this is not a key", id="random-string"), + pytest.param( + "not-a-key-type AAAAB3NzaC1yc2EAAAADAQABAAABAQ", + id="malformed-key-type", + ), + ], + ) + def test_invalid(self, value): + """Non-OpenSSH strings should return False.""" + assert certs.is_openssh_formatted(value) is False + + +class TestSanitizeOpensshKey: + """Test sanitize_openssh_key() function.""" + + @pytest.mark.parametrize( + "key, expected", + [ + pytest.param( + "ssh-rsa AAAA\r\nBBBB user@host", + "ssh-rsa AAAABBBB user@host", + id="removes-embedded-crlf", + ), + pytest.param( + "ssh-rsa AAAABBBB user@host\r\n", + "ssh-rsa AAAABBBB user@host", + id="strips-trailing-crlf", + ), + pytest.param( + "ssh-rsa AAAABBBB user@host", + "ssh-rsa AAAABBBB user@host", + id="clean-key-unchanged", + ), + pytest.param( + "ssh-rsa AA\r\nAA\r\nBBBB user@host", + "ssh-rsa AAAABBBB user@host", + id="multiple-crlf-sequences", + ), + ], + ) + def test_sanitize(self, key, expected): + """Sanitized keys should have CRLF sequences removed.""" + assert certs.sanitize_openssh_key(key) == expected + + +class TestIsX509Certificate: + """Test is_x509_certificate() function.""" + + @pytest.mark.parametrize( + "value", + [ + pytest.param(_VALID_RSA_KEY, id="openssh-key"), + pytest.param("", id="empty-string"), + pytest.param("this is not a certificate", id="random-string"), + pytest.param( + "-----BEGIN CERTIFICATE-----", id="only-begin-marker" + ), + pytest.param("-----END CERTIFICATE-----", id="only-end-marker"), + ], + ) + def test_invalid(self, value): + """Non-certificate inputs should return False.""" + assert certs.is_x509_certificate(value) is False + + @mock.patch("cloudinit.sources.azure.certs.subp.subp") + def test_invalid_certificate_content(self, m_subp): + """Certificate with invalid content should return False.""" + m_subp.side_effect = subp.ProcessExecutionError( + "unable to load certificate" + ) + + assert certs.is_x509_certificate(_INVALID_X509_CERT) is False + m_subp.assert_called_once() + + @pytest.mark.skipif( + shutil.which("openssl") is None, reason="openssl not available" + ) + @pytest.mark.allow_subp_for("openssl") + def test_valid_certificate_integration(self, cert_data): + """Integration test: Actually validate certificate with openssl.""" + assert certs.is_x509_certificate(cert_data) is True + + @pytest.mark.skipif( + shutil.which("openssl") is None, reason="openssl not available" + ) + @pytest.mark.allow_subp_for("openssl") + def test_certificate_with_extra_whitespace_integration(self, cert_data): + """Integration test: Certificate with extra whitespace validates.""" + assert certs.is_x509_certificate("\n\n" + cert_data + "\n\n") is True + + @pytest.mark.skipif( + shutil.which("openssl") is None, reason="openssl not available" + ) + @pytest.mark.allow_subp_for("openssl") + def test_invalid_certificate_integration(self): + """Integration test: Reject invalid certificate with real openssl.""" + assert certs.is_x509_certificate(_INVALID_X509_CERT) is False + + +class TestExtractX509Certificates: + """Test extract_x509_certificates() function.""" + + @pytest.mark.parametrize( + "data", + [ + pytest.param("", id="empty-string"), + pytest.param(None, id="none"), + pytest.param( + "this is just some random text\nwith no certificate", + id="no-cert-markers", + ), + ], + ) + def test_returns_empty_list(self, data): + """Data with no certificates should return empty list.""" + assert certs.extract_x509_certificates(data) == [] + + @mock.patch("cloudinit.sources.azure.certs.is_x509_certificate") + def test_extracts_all_valid_certificates(self, m_is_x509): + """Should extract and return all valid certificates.""" + cert1 = dedent( + """\ + -----BEGIN CERTIFICATE----- + CERT1DATA + -----END CERTIFICATE----- + """ + ) + cert2 = dedent( + """\ + -----BEGIN CERTIFICATE----- + CERT2DATA + -----END CERTIFICATE----- + """ + ) + bundle = cert1 + "\n" + cert2 + + m_is_x509.return_value = True + + result = certs.extract_x509_certificates(bundle) + + assert result == [cert1.strip(), cert2.strip()] + + @mock.patch("cloudinit.sources.azure.certs.is_x509_certificate") + def test_skips_private_keys(self, m_is_x509): + """Should skip private keys and extract certificate.""" + private_key = dedent( + """\ + -----BEGIN PRIVATE KEY----- + PRIVATEKEYDATA + -----END PRIVATE KEY----- + """ + ) + certificate = dedent( + """\ + -----BEGIN CERTIFICATE----- + CERTDATA + -----END CERTIFICATE----- + """ + ) + bundle = private_key + "\n" + certificate + + m_is_x509.return_value = True + + result = certs.extract_x509_certificates(bundle) + + assert result == [certificate.strip()] + + @mock.patch("cloudinit.sources.azure.certs.is_x509_certificate") + def test_skips_invalid_certs(self, m_is_x509): + """Should skip invalid cert and return only valid ones.""" + invalid_cert = dedent( + """\ + -----BEGIN CERTIFICATE----- + INVALID + -----END CERTIFICATE----- + """ + ) + valid_cert = dedent( + """\ + -----BEGIN CERTIFICATE----- + VALID + -----END CERTIFICATE----- + """ + ) + bundle = invalid_cert + "\n" + valid_cert + + m_is_x509.side_effect = [False, True] + + result = certs.extract_x509_certificates(bundle) + + assert result == [valid_cert.strip()] + + @pytest.mark.skipif( + shutil.which("openssl") is None, reason="openssl not available" + ) + @pytest.mark.allow_subp_for("openssl") + def test_extraction_from_mixed_bundle_integration(self, cert_data): + """Integration test: Extract cert from bundle with private key.""" + private_key = dedent( + """\ + -----BEGIN PRIVATE KEY----- + MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDExample + -----END PRIVATE KEY----- + """ + ) + bundle = private_key + "\n" + cert_data + + result = certs.extract_x509_certificates(bundle) + + assert result == [cert_data.strip()] + + +class TestConvertX509ToOpenssh: + """Test convert_x509_to_openssh() function.""" + + @mock.patch("cloudinit.sources.azure.certs.subp.subp") + def test_conversion_with_mocked_commands(self, m_subp): + """Test basic conversion flow with mocked subp calls.""" + cert = dedent( + """\ + -----BEGIN CERTIFICATE----- + MIIB+TCCAeOgAwIBAgIBATANBgkqhkiG9w0BAQUFADAWMRQwEgYDVQQDDAtSb290 + -----END CERTIFICATE----- + """ + ) + pubkey = ( + "-----BEGIN PUBLIC KEY-----\nMIIB...\n-----END PUBLIC KEY-----" + ) + expected_ssh_key = "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQ..." + + m_subp.side_effect = [ + (pubkey, ""), + (expected_ssh_key, ""), + ] + + result = certs.convert_x509_to_openssh(cert) + + assert result == expected_ssh_key + assert m_subp.call_count == 2 + + openssl_call = m_subp.call_args_list[0] + assert openssl_call[0][0] == ["openssl", "x509", "-noout", "-pubkey"] + assert openssl_call[1]["data"] == cert + + keygen_call = m_subp.call_args_list[1] + assert keygen_call[0][0] == [ + "ssh-keygen", + "-i", + "-m", + "PKCS8", + "-f", + "/dev/stdin", + ] + assert keygen_call[1]["data"] == pubkey + + @mock.patch("cloudinit.sources.azure.certs.subp.subp") + def test_openssl_failure_raises_exception(self, m_subp): + """Test that openssl failure raises ProcessExecutionError.""" + cert = ( + "-----BEGIN CERTIFICATE-----\nINVALID\n-----END CERTIFICATE-----" + ) + + m_subp.side_effect = subp.ProcessExecutionError("openssl failed") + + with pytest.raises(subp.ProcessExecutionError) as exc_info: + certs.convert_x509_to_openssh(cert) + + assert "openssl failed" in str(exc_info.value) + + @mock.patch("cloudinit.sources.azure.certs.subp.subp") + def test_ssh_keygen_failure_raises_exception(self, m_subp): + """Test that ssh-keygen failure raises ProcessExecutionError.""" + cert = "-----BEGIN CERTIFICATE-----\nVALID\n-----END CERTIFICATE-----" + pubkey = ( + "-----BEGIN PUBLIC KEY-----\nMIIB...\n-----END PUBLIC KEY-----" + ) + + m_subp.side_effect = [ + (pubkey, ""), + subp.ProcessExecutionError("ssh-keygen failed"), + ] + + with pytest.raises(subp.ProcessExecutionError) as exc_info: + certs.convert_x509_to_openssh(cert) + + assert "ssh-keygen failed" in str(exc_info.value) + + @pytest.mark.skipif( + shutil.which("openssl") is None or shutil.which("ssh-keygen") is None, + reason="openssl or ssh-keygen not available", + ) + @pytest.mark.allow_subp_for("openssl", "ssh-keygen") + def test_conversion_integration(self, cert_data, data_file_path): + """Integration test: Convert certificate with real commands.""" + key_file = data_file_path("pubkey_extract_ssh_key") + + if not key_file.exists(): + pytest.skip("Test data files not found") + + expected_key = key_file.read_text().strip() + + result = certs.convert_x509_to_openssh(cert_data) + assert result.strip() == expected_key diff --git a/tests/unittests/sources/test_azure.py b/tests/unittests/sources/test_azure.py index 807acdbdd71..d344575991d 100644 --- a/tests/unittests/sources/test_azure.py +++ b/tests/unittests/sources/test_azure.py @@ -25,7 +25,7 @@ from cloudinit.net import dhcp, ephemeral from cloudinit.sources import UNSET from cloudinit.sources import DataSourceAzure as dsaz -from cloudinit.sources.azure import errors, identity, imds +from cloudinit.sources.azure import certs, errors, identity, imds from cloudinit.sources.helpers import netlink from cloudinit.util import ( MountFailedError, @@ -2342,15 +2342,15 @@ def test_get_public_ssh_keys_with_imds(self, m_parse_certificates, get_ds): def test_key_without_crlf_valid(self): test_key = "ssh-rsa somerandomkeystuff some comment" - assert True is dsaz._key_is_openssh_formatted(test_key) + assert True is certs.is_openssh_formatted(test_key) - def test_key_with_crlf_invalid(self): + def test_key_with_crlf_sanitized(self): test_key = "ssh-rsa someran\r\ndomkeystuff some comment" - assert False is dsaz._key_is_openssh_formatted(test_key) + assert True is certs.is_openssh_formatted(test_key) def test_key_endswith_crlf_valid(self): test_key = "ssh-rsa somerandomkeystuff some comment\r\n" - assert True is dsaz._key_is_openssh_formatted(test_key) + assert True is certs.is_openssh_formatted(test_key) @mock.patch( "cloudinit.sources.helpers.azure.OpenSSLManager.parse_certificates" diff --git a/tests/unittests/sources/test_azure_helper.py b/tests/unittests/sources/test_azure_helper.py index 643637753f1..0729bc7b354 100644 --- a/tests/unittests/sources/test_azure_helper.py +++ b/tests/unittests/sources/test_azure_helper.py @@ -556,6 +556,62 @@ def test_parse_certificates(self, mock_decrypt_certs): for fp in fingerprints: assert fp in keys_by_fp + def test_parse_certificates_processes_all_certificates(self): + pem_bundle = ( + "header\r\n" + "-----BEGIN CERTIFICATE-----\r\n" + "CERTDATA1\r\n" + "-----END CERTIFICATE-----\r\n" + "-----BEGIN PRIVATE KEY-----\r\n" + "KEYDATA\r\n" + "-----END PRIVATE KEY-----\r\n" + "-----BEGIN CERTIFICATE-----\r\n" + "CERTDATA2\r\n" + "-----END CERTIFICATE-----\r\n" + ) + cert1 = ( + "-----BEGIN CERTIFICATE-----\r\n" + "CERTDATA1\r\n" + "-----END CERTIFICATE-----" + ) + cert2 = ( + "-----BEGIN CERTIFICATE-----\r\n" + "CERTDATA2\r\n" + "-----END CERTIFICATE-----" + ) + with mock.patch.object( + azure_helper.OpenSSLManager, + "generate_certificate", + ), mock.patch.object( + azure_helper.OpenSSLManager, + "_decrypt_certs_from_xml", + return_value=pem_bundle, + ) as m_decrypt, mock.patch.object( + azure_helper.OpenSSLManager, + "_get_ssh_key_from_cert", + side_effect=["ssh-cert-1", "ssh-cert-2"], + ) as m_get_key, mock.patch.object( + azure_helper.OpenSSLManager, + "_get_fingerprint_from_cert", + side_effect=["fp1", "fp2"], + ) as m_get_fp, mock.patch( + "cloudinit.sources.helpers.azure.certs.extract_x509_certificates", + return_value=[cert1, cert2], + ): + sslmgr = azure_helper.OpenSSLManager() + result = sslmgr.parse_certificates("fake-xml") + + assert result == {"fp1": "ssh-cert-1", "fp2": "ssh-cert-2"} + m_decrypt.assert_called_once_with("fake-xml") + assert m_get_key.call_args_list == [ + mock.call(cert1), + mock.call(cert2), + ] + assert m_get_fp.call_args_list == [ + mock.call(cert1), + mock.call(cert2), + ] + class TestGoalStateHealthReporter: maxDiff = None diff --git a/tests/unittests/sources/test_opennebula.py b/tests/unittests/sources/test_opennebula.py index 1c7e48691dd..7cd100d8872 100644 --- a/tests/unittests/sources/test_opennebula.py +++ b/tests/unittests/sources/test_opennebula.py @@ -5,9 +5,11 @@ import pwd from unittest import mock +import jsonschema import pytest from cloudinit import atomic_helper +from cloudinit.config.schema import SchemaType, get_schema from cloudinit.sources import DataSourceOpenNebula as ds from tests.unittests.helpers import populate_dir @@ -370,9 +372,24 @@ def my_devs_with(criteria): @mock.patch(DS_PATH + ".net.get_interfaces_by_mac", mock.Mock(return_value={})) class TestOpenNebulaNetwork: - system_nics = ("eth0", "ens3") + @pytest.fixture(autouse=True) + def _validate_gen_conf_schema(self, monkeypatch): + """Wrap gen_conf() to assert schema validity on every test.""" + schema = get_schema(SchemaType.NETWORK_CONFIG_V2) + assert schema, "network-config-v2 schema must not be empty" + validator_cls = jsonschema.validators.validator_for(schema) + validator = validator_cls(schema) + original = ds.OpenNebulaNetwork.gen_conf + + def validated(self_inner): + result = original(self_inner) + validator.validate(result) + return result + + monkeypatch.setattr(ds.OpenNebulaNetwork, "gen_conf", validated) + def test_context_devname(self): """Verify context_devname correctly returns mac and name.""" context = { diff --git a/tests/unittests/sources/test_oracle.py b/tests/unittests/sources/test_oracle.py index 4b39f215566..c69f314e2f3 100644 --- a/tests/unittests/sources/test_oracle.py +++ b/tests/unittests/sources/test_oracle.py @@ -381,6 +381,48 @@ def test_ds_detect(self, dmi_data, platform_viable): m_read_dmi_data.assert_has_calls([mock.call("chassis-asset-tag")]) +class TestIbftHasIscsiBootTarget: + @pytest.mark.parametrize( + "flags_contents, is_iscsi_root", + [ + # Valid and firmware-boot-selected target is an iSCSI root. + (["3"], True), + # Valid but not boot-selected is not. + (["1"], False), + # Neither valid nor boot-selected is not. + (["0"], False), + # Boot-selected but not valid is not. + (["2"], False), + # Any valid and boot-selected target among several wins. + (["0", "3"], True), + (["1", "2"], False), + # Malformed flag contents are ignored. + (["garbage"], False), + # No iBFT targets present. + ([], False), + ], + ) + def test_flag_values(self, flags_contents, is_iscsi_root, mocker): + paths = [ + f"/sys/firmware/ibft/target{i}/flags" + for i in range(len(flags_contents)) + ] + mocker.patch(DS_PATH + ".glob.glob", return_value=paths) + mocker.patch( + DS_PATH + ".util.load_text_file", side_effect=flags_contents + ) + assert is_iscsi_root == oracle._ibft_has_iscsi_boot_target() + + @pytest.mark.parametrize("error", [FileNotFoundError, PermissionError]) + def test_unreadable_flags_are_skipped(self, error, mocker): + mocker.patch( + DS_PATH + ".glob.glob", + return_value=["/sys/firmware/ibft/target0/flags"], + ) + mocker.patch(DS_PATH + ".util.load_text_file", side_effect=error) + assert not oracle._ibft_has_iscsi_boot_target() + + @pytest.mark.is_iscsi(False) @mock.patch( "cloudinit.net.is_openvswitch_internal_interface", @@ -1412,6 +1454,50 @@ def test_network_config_cached(self, m_get_interfaces_by_mac, oracle_ds): oracle_ds.network_config # pylint: disable=pointless-statement assert 1 == oracle_ds._get_iscsi_config.call_count + @pytest.mark.is_iscsi(True) + def test_keep_configuration_set_from_iscsi_klibc( + self, m_get_interfaces_by_mac, oracle_ds + ): + """iSCSI root config from initramfs marks the primary NIC critical.""" + netcfg = oracle_ds.network_config + assert netcfg["config"][0]["keep_configuration"] is True + + @pytest.mark.is_iscsi(True) + def test_keep_configuration_set_from_imds_fallback( + self, m_get_interfaces_by_mac, oracle_ds, mocker + ): + """iSCSI root with no klibc config (dracut) still marks the + primary NIC critical when config comes from IMDS.""" + m_get_interfaces_by_mac.return_value = { + "02:00:17:05:d1:db": "ens3", + "00:00:17:02:2b:b1": "ens4", + } + mocker.patch.object( + oracle_ds._network_config_source, + "is_applicable", + return_value=False, + ) + oracle_ds._vnics_data = json.loads(OPC_VM_SECONDARY_VNIC_RESPONSE) + + netcfg = oracle_ds.network_config + + assert netcfg["config"][0]["keep_configuration"] is True + + @pytest.mark.is_iscsi(False) + def test_keep_configuration_not_set_without_iscsi( + self, m_get_interfaces_by_mac, oracle_ds + ): + """Non-iSCSI instances do not mark the primary NIC critical.""" + m_get_interfaces_by_mac.return_value = { + "02:00:17:05:d1:db": "ens3", + "00:00:17:02:2b:b1": "ens4", + } + oracle_ds._vnics_data = json.loads(OPC_VM_SECONDARY_VNIC_RESPONSE) + + netcfg = oracle_ds.network_config + + assert "keep_configuration" not in netcfg["config"][0] + @pytest.mark.parametrize( "configure_secondary_nics,is_iscsi,expected_set_primary", [ diff --git a/tests/unittests/test_builtin_handlers.py b/tests/unittests/test_builtin_handlers.py index cd545479729..a23ed55bd75 100644 --- a/tests/unittests/test_builtin_handlers.py +++ b/tests/unittests/test_builtin_handlers.py @@ -388,6 +388,23 @@ def test_render_jinja_payload_replaces_missing_variables_and_warns( ) assert expected_log in caplog.text + def test_render_jinja_payload_blocks_unsafe_attribute_access(self, caplog): + payload = ( + "## template: jinja\n" + "{{ ''.__class__.__mro__[1].__subclasses__()[:3] }}" + ) + + assert ( + render_jinja_payload( + payload=payload, + payload_fn="myfile", + instance_data={"v1": {"hostname": "foo"}}, + ) + is None + ) + assert "Ignoring jinja template for myfile" in caplog.text + assert "__class__" in caplog.text + class TestShellScriptByFrequencyHandlers: @pytest.fixture(autouse=True) diff --git a/tests/unittests/test_templating.py b/tests/unittests/test_templating.py index a5aab1f9a03..be3af247c19 100644 --- a/tests/unittests/test_templating.py +++ b/tests/unittests/test_templating.py @@ -7,6 +7,7 @@ import textwrap import pytest +from jinja2.exceptions import SecurityError from cloudinit import templater from cloudinit.templater import JinjaSyntaxParsingException @@ -148,6 +149,15 @@ def test_jinja_do_extension_render_to_string(self): == expected_result ) + def test_jinja_blocks_unsafe_attribute_access(self): + template = self.add_header( + "jinja", + "{{ ''.__class__.__mro__[1].__subclasses__()[:3] }}", + ) + + with pytest.raises(SecurityError): + templater.render_string(template, {}) + class TestJinjaSyntaxParsingException: def test_jinja_syntax_parsing_exception_message(self): diff --git a/tests/unittests/test_util.py b/tests/unittests/test_util.py index fe24962e19b..7cdcaf484fa 100644 --- a/tests/unittests/test_util.py +++ b/tests/unittests/test_util.py @@ -2145,6 +2145,44 @@ def test_none_returns_default(self): ] +class TestFstabEscaping: + @pytest.mark.parametrize( + "raw, escaped", + [ + ("/mnt/Cdrom Drive", "/mnt/Cdrom\\040Drive"), + ("/mnt/a\tb", "/mnt/a\\011b"), + ("/mnt/a\nb", "/mnt/a\\012b"), + ("/mnt/a\\b", "/mnt/a\\134b"), + # backslash escaped first so existing escapes aren't doubled + ("/mnt/a \\b", "/mnt/a\\040\\134b"), + ("/mnt/plain", "/mnt/plain"), + ], + ) + def test_escape_unescape_roundtrip(self, raw, escaped): + assert util.escape_fstab_field(raw) == escaped + assert util.unescape_fstab_field(escaped) == raw + + +class TestMounts: + def test_mounts_unescapes_mountpoint(self, mocker): + """mounts() octal-unescapes fs_file via unescape_fstab_field. + + Exercises the ``mount`` command fallback (no /proc/mounts), where an + escaped space (octal 040) appears in the mount point of the output. + """ + # No /proc/mounts, so mounts() parses the `mount` command output. + mocker.patch(M_PATH + "os.path.exists", return_value=False) + mocker.patch( + M_PATH + "subp.subp", + return_value=SubpResult( + "/dev/sr0 on /mnt/Cdrom\\040Drive (ufs, local, journaled)\n", + "", + ), + ) + result = util.mounts() + assert result["/dev/sr0"]["mountpoint"] == "/mnt/Cdrom Drive" + + class TestMountinfoParsing: def test_invalid_mountinfo(self): line = (