diff --git a/implementations/python/.DS_Store b/implementations/python/.DS_Store new file mode 100644 index 00000000..c78a9a05 Binary files /dev/null and b/implementations/python/.DS_Store differ diff --git a/implementations/python/packages/raes_backend_libvirt/guest_appliance.py b/implementations/python/packages/raes_backend_libvirt/guest_appliance.py index 30951eee..59d8fef2 100644 --- a/implementations/python/packages/raes_backend_libvirt/guest_appliance.py +++ b/implementations/python/packages/raes_backend_libvirt/guest_appliance.py @@ -19,7 +19,7 @@ from dataclasses import dataclass from pathlib import Path -from .techvault_appliance import _cpio_newc, _shell_quote +from .techvault_appliance import _cpio_newc, _interface_case_lines, _shell_quote _APPLETS = ( "sh", "mount", "mdev", "ip", "ifconfig", "sleep", "cat", "hostname", "printf", "echo", @@ -132,13 +132,7 @@ def _init_script(domain: Mapping[str, object]) -> str: for interface in _as_sequence(domain.get("interfaces")): if not isinstance(interface, Mapping): continue - lines.extend( - [ - f" {interface.get('mac')})", - f' ip addr add {interface.get("ip")}/{interface.get("cidr_prefix")} dev "$iface"', - " ;;", - ] - ) + lines.extend(_interface_case_lines(interface)) lines.extend([" esac", "done"]) lines.extend(_REALIZE_SNIPPET) lines.append("sleep 1") diff --git a/implementations/python/packages/raes_backend_libvirt/techvault_appliance.py b/implementations/python/packages/raes_backend_libvirt/techvault_appliance.py index 6773ffc3..b51b1318 100644 --- a/implementations/python/packages/raes_backend_libvirt/techvault_appliance.py +++ b/implementations/python/packages/raes_backend_libvirt/techvault_appliance.py @@ -3,8 +3,10 @@ from __future__ import annotations import gzip +import ipaddress import json import os +import re import shutil import stat import subprocess @@ -91,22 +93,20 @@ def _init_script(domain: Mapping[str, object]) -> str: for interface in _as_sequence(domain.get("interfaces")): if not isinstance(interface, Mapping): continue - lines.extend( - [ - f" {interface.get('mac')})", - f' ip addr add {interface.get("ip")}/{interface.get("cidr_prefix")} dev "$iface"', - " ;;", - ] - ) + lines.extend(_interface_case_lines(interface)) lines.extend([" esac", "done"]) lines.extend(["while true; do sleep 3600; done", ""]) return "\n".join(lines) def _cpio_newc(root: Path) -> bytes: + paths = _cpio_paths(root) + for path in paths: + if "\n" in path: + raise ValueError(f"initramfs member path contains a newline: {path!r}") proc = subprocess.run( ["cpio", "-o", "-H", "newc", "--quiet"], - input=("\n".join(_cpio_paths(root)) + "\n").encode(), + input=("\n".join(paths) + "\n").encode(), cwd=root, capture_output=True, check=False, @@ -124,5 +124,58 @@ def _as_sequence(value: object) -> Sequence[object]: return value if isinstance(value, list | tuple) else () +_MAC_RE = re.compile(r"\A[0-9A-Fa-f]{2}(?::[0-9A-Fa-f]{2}){5}\Z") + + +def _interface_case_lines(interface: Mapping[str, object]) -> list[str]: + """Render one shell-safe ``case`` arm configuring an interface address. + + ``mac``/``ip``/``cidr_prefix`` are interpolated into a root-run guest init + script. Each is validated to its structural shape and rejected when + malformed, then quoted as defense in depth: a field that is not the shape it + claims to be (a ``mac`` that is not a MAC) is a bug in the plan, not text to + escape into a command. + """ + + mac = _validated_mac(interface.get("mac")) + address = _validated_interface_address(interface.get("ip"), interface.get("cidr_prefix")) + return [ + f" {_shell_quote(mac)})", + f' ip addr add {_shell_quote(address)} dev "$iface"', + " ;;", + ] + + +def _validated_mac(value: object) -> str: + if isinstance(value, str) and _MAC_RE.match(value): + return value + raise ValueError(f"interface mac is not a MAC address: {value!r}") + + +def _validated_interface_address(ip: object, cidr_prefix: object) -> str: + if not isinstance(ip, str): + raise ValueError(f"interface ip is not an IP address: {ip!r}") + try: + parsed = ipaddress.ip_address(ip) + except ValueError as error: + raise ValueError(f"interface ip is not an IP address: {ip!r}") from error + return f"{ip}/{_validated_cidr_prefix(cidr_prefix, parsed.max_prefixlen)}" + + +def _validated_cidr_prefix(value: object, max_prefix: int) -> int: + if isinstance(value, bool): + raise ValueError(f"interface cidr_prefix is not an integer: {value!r}") + # ``isdecimal`` rather than ``isdigit``: the latter also accepts characters + # like "²" that ``int`` then refuses, leaking a bare interpreter message + # instead of this function's. + if isinstance(value, str) and value.isdecimal(): + value = int(value) + if not isinstance(value, int): + raise ValueError(f"interface cidr_prefix is not an integer: {value!r}") + if not 0 <= value <= max_prefix: + raise ValueError(f"interface cidr_prefix is out of range 0..{max_prefix}: {value!r}") + return value + + def _shell_quote(value: str) -> str: return "'" + value.replace("'", "'\"'\"'") + "'" diff --git a/implementations/python/tests/libvirt_interface_fixtures.py b/implementations/python/tests/libvirt_interface_fixtures.py new file mode 100644 index 00000000..d5b8d2d9 --- /dev/null +++ b/implementations/python/tests/libvirt_interface_fixtures.py @@ -0,0 +1,39 @@ +"""Shared interface fixtures for the libvirt guest init-script generators. + +`techvault_appliance` and `guest_appliance` build the same root-run init script +and share one validated interface renderer, so their tests assert the same +guarantees against two entry points. The fixtures live here so neither test +module duplicates the other. +""" + +from __future__ import annotations + +VALID_INTERFACE = {"mac": "52:54:00:00:00:01", "ip": "192.0.2.10", "cidr_prefix": 24} + +# Each case pairs a hostile or ill-typed field with the message fragment the +# generator must raise. Shell metacharacters must be refused outright rather +# than escaped: a field that is not the shape it claims to be is a bug in the +# plan, not text to quote into a root-run command. +HOSTILE_INTERFACE_CASES = ( + ({**VALID_INTERFACE, "mac": "aa:bb:cc:dd:ee:ff) ; rm -rf /outside #"}, "mac is not a MAC"), + ({**VALID_INTERFACE, "ip": "192.0.2.10$(touch /pwned)"}, "ip is not an IP"), + ({**VALID_INTERFACE, "ip": "`reboot`"}, "ip is not an IP"), + ({**VALID_INTERFACE, "cidr_prefix": "24; rm -rf /"}, "cidr_prefix is not an integer"), + ({**VALID_INTERFACE, "cidr_prefix": 33}, "cidr_prefix is out of range"), + ({**VALID_INTERFACE, "ip": 3221225994}, "ip is not an IP"), + ({**VALID_INTERFACE, "ip": None}, "ip is not an IP"), + ({**VALID_INTERFACE, "cidr_prefix": True}, "cidr_prefix is not an integer"), + ({**VALID_INTERFACE, "cidr_prefix": "²"}, "cidr_prefix is not an integer"), + ({**VALID_INTERFACE, "cidr_prefix": 24.5}, "cidr_prefix is not an integer"), +) + +QUOTED_MAC_ARM = " '52:54:00:00:00:01')" +QUOTED_ADDRESS_COMMAND = "ip addr add '192.0.2.10/24' dev \"$iface\"" + + +def domain_with_interface(**interface: object) -> dict[str, object]: + return {"name": "webapp", "interfaces": [interface]} + + +def domain_with_malformed_entry() -> dict[str, object]: + return {"name": "webapp", "interfaces": ["not-a-mapping", VALID_INTERFACE]} diff --git a/implementations/python/tests/test_libvirt_backend_techvault_native.py b/implementations/python/tests/test_libvirt_backend_techvault_native.py index 8406e48b..543ffd05 100644 --- a/implementations/python/tests/test_libvirt_backend_techvault_native.py +++ b/implementations/python/tests/test_libvirt_backend_techvault_native.py @@ -9,12 +9,21 @@ from pathlib import Path import pytest +from libvirt_interface_fixtures import ( + HOSTILE_INTERFACE_CASES, + QUOTED_ADDRESS_COMMAND, + QUOTED_MAC_ARM, + VALID_INTERFACE, + domain_with_interface, + domain_with_malformed_entry, +) from paths import EXAMPLES_DIR from raes import parse_sdl from raes_backend_libvirt import create_libvirt_target from raes_backend_libvirt.cloudinit import CloudInitSpec, CloudInitUser from raes_backend_libvirt.driver import DomainSpec, NetworkAcl, NetworkSpec, ServiceSpec from raes_backend_libvirt.envelopes import load_libvirt_realization_envelope +from raes_backend_libvirt.techvault_appliance import _cpio_newc, _init_script from raes_backend_libvirt.techvault_native import ( BusyboxInitramfsBuilder, ProbeResult, @@ -923,3 +932,43 @@ def test_busybox_initramfs_builder_writes_gzip_cpio(tmp_path): assert target.read_bytes().startswith(b"\x1f\x8b") assert target.stat().st_size > 1000 assert b"httpd -p" not in gzip.decompress(target.read_bytes()) + + +def test_init_script_accepts_a_decimal_string_cidr_prefix(): + script = _init_script(domain_with_interface(**{**VALID_INTERFACE, "cidr_prefix": "24"})) + + assert QUOTED_ADDRESS_COMMAND in script + + +def test_init_script_quotes_valid_interface_addressing(): + script = _init_script(domain_with_interface(**VALID_INTERFACE)) + + assert QUOTED_MAC_ARM in script + assert QUOTED_ADDRESS_COMMAND in script + + +@pytest.mark.parametrize(("interface", "match"), HOSTILE_INTERFACE_CASES) +def test_init_script_rejects_hostile_interface_fields_before_scripting(interface, match): + # A field that is not the shape it claims to be aborts script generation, so + # no attacker-controlled shell can reach the root-run guest init script. + # The domain is built outside the block so the generator is the only call + # inside it that can raise. + domain = domain_with_interface(**interface) + + with pytest.raises(ValueError, match=match): + _init_script(domain) + + +def test_init_script_skips_a_malformed_interface_entry_and_renders_the_rest(): + script = _init_script(domain_with_malformed_entry()) + + assert "not-a-mapping" not in script + assert QUOTED_MAC_ARM in script + + +def test_cpio_newc_rejects_newline_in_member_path(tmp_path): + (tmp_path / "bin").mkdir() + (tmp_path / "evil\nname").write_text("payload", encoding="utf-8") + + with pytest.raises(ValueError, match="newline"): + _cpio_newc(tmp_path) diff --git a/implementations/python/tests/test_libvirt_guest_appliance_init_script.py b/implementations/python/tests/test_libvirt_guest_appliance_init_script.py new file mode 100644 index 00000000..b996ffa4 --- /dev/null +++ b/implementations/python/tests/test_libvirt_guest_appliance_init_script.py @@ -0,0 +1,50 @@ +"""Guest-appliance init script must reject hostile interface fields. + +`guest_appliance` builds the same root-run init script as `techvault_appliance` +and shares its interface-rendering helper. Its own coverage lives in +`test_libvirt_backend_guest_certified.py`, which needs a static BusyBox and +`cpio` and therefore skips or fails on hosts without them; these tests exercise +the script generator directly so the second injection site stays covered +everywhere. +""" + +from __future__ import annotations + +import pytest +from libvirt_interface_fixtures import ( + HOSTILE_INTERFACE_CASES, + QUOTED_ADDRESS_COMMAND, + QUOTED_MAC_ARM, + VALID_INTERFACE, + domain_with_interface, + domain_with_malformed_entry, +) +from raes_backend_libvirt.guest_appliance import _init_script + + +def test_guest_init_script_quotes_valid_interface_addressing(): + script = _init_script(domain_with_interface(**VALID_INTERFACE)) + + assert QUOTED_MAC_ARM in script + assert QUOTED_ADDRESS_COMMAND in script + + +@pytest.mark.parametrize(("interface", "match"), HOSTILE_INTERFACE_CASES) +def test_guest_init_script_rejects_hostile_interface_fields(interface, match): + domain = domain_with_interface(**interface) + + with pytest.raises(ValueError, match=match): + _init_script(domain) + + +def test_guest_init_script_quotes_the_hostname(): + script = _init_script({"name": "web; rm -rf /", "interfaces": []}) + + assert "hostname 'web; rm -rf /'" in script + + +def test_guest_init_script_skips_malformed_interface_entries(): + script = _init_script(domain_with_malformed_entry()) + + assert QUOTED_MAC_ARM in script + assert "not-a-mapping" not in script