Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Binary file added implementations/python/.DS_Store
Binary file not shown.
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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")
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,10 @@
from __future__ import annotations

import gzip
import ipaddress
import json
import os
import re
import shutil
import stat
import subprocess
Expand Down Expand Up @@ -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,
Expand All @@ -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("'", "'\"'\"'") + "'"
39 changes: 39 additions & 0 deletions implementations/python/tests/libvirt_interface_fixtures.py
Original file line number Diff line number Diff line change
@@ -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]}
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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)
Original file line number Diff line number Diff line change
@@ -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