kstyagi
d8a25209b9
- Introduced a new method `installed_kernel` in `ImageStore` to handle kernel verification and checksum validation. - Updated `Installer` to include additional packages: `nftables` and `conntrack`. - Refined `LifecycleService` to improve error handling and state management during VM creation, starting, stopping, and deletion. - Enhanced port management with new methods for setting and syncing ports, including validation for port rules. - Improved error handling in `StateStore` to raise specific `NotFoundError` and `ConflictError` exceptions. - Updated network management to ensure proper TAP device handling and firewall integration. - Enhanced API responses in `vms.py` to include policy status for better client-side handling. - Added validation for VM names and ports in `validation.py` to enforce stricter rules and prevent conflicts.
150 lines
7.1 KiB
Python
150 lines
7.1 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from uvm.domain import PortRule
|
|
from uvm.errors import ValidationError
|
|
from uvm.validation import (
|
|
parse_cpu, parse_name, parse_password, parse_ports, parse_ram, parse_timeout, parse_username,
|
|
)
|
|
|
|
|
|
class ValidationTests(unittest.TestCase):
|
|
def test_parse_ram_uses_mib_by_default(self) -> None:
|
|
self.assertEqual(parse_ram("512"), 512)
|
|
self.assertEqual(parse_ram("1G"), 1024)
|
|
self.assertEqual(parse_ram("512MiB"), 512)
|
|
self.assertEqual(parse_ram("134217728B"), 128)
|
|
|
|
def test_parse_ram_rejects_small_and_invalid_values(self) -> None:
|
|
for value in ("127M", "nonsense", "-1G", "1T"):
|
|
with self.subTest(value=value):
|
|
with self.assertRaises(ValidationError):
|
|
parse_ram(value)
|
|
|
|
def test_parse_cpu_requires_a_positive_finite_value(self) -> None:
|
|
self.assertEqual(parse_cpu("0.5"), 0.5)
|
|
for value in ("0", "-1", "nan", "inf", "cpu"):
|
|
with self.subTest(value=value):
|
|
with self.assertRaises(ValidationError):
|
|
parse_cpu(value)
|
|
|
|
def test_guest_username_is_validated(self) -> None:
|
|
self.assertEqual(parse_username("root"), "root")
|
|
self.assertEqual(parse_username("app-user"), "app-user")
|
|
for value in ("", "Root", "9user", "user name", "a" * 33):
|
|
with self.subTest(value=value):
|
|
with self.assertRaises(ValidationError):
|
|
parse_username(value)
|
|
|
|
def test_guest_password_rejects_empty_control_or_oversized_values(self) -> None:
|
|
self.assertEqual(
|
|
parse_password("correct horse battery staple"),
|
|
"correct horse battery staple",
|
|
)
|
|
for value in ("", "line\nbreak", "tab\tvalue", "a" * 129):
|
|
with self.subTest(value=value):
|
|
with self.assertRaises(ValidationError):
|
|
parse_password(value)
|
|
|
|
def test_names_accept_full_lowercase_names_up_to_63_characters(self) -> None:
|
|
for name in ("a", "app-server", "worker2", "a" * 63, "vm-test", "vm-" + "g" * 32):
|
|
with self.subTest(name=name):
|
|
self.assertEqual(parse_name(name), name)
|
|
|
|
def test_names_reject_invalid_format_case_and_length(self) -> None:
|
|
for name in (
|
|
"", "App", "APP", "9app", "-app", "app_name", "app.name", "app/name",
|
|
" app", "app ", "app\n", "a" * 64, "caf\u00e9", "app;id",
|
|
):
|
|
with self.subTest(name=name), self.assertRaises(ValidationError):
|
|
parse_name(name)
|
|
|
|
def test_names_cannot_impersonate_a_generated_vm_id(self) -> None:
|
|
for suffix in ("0" * 32, "0123456789abcdef" * 2, "f" * 32):
|
|
with self.subTest(suffix=suffix):
|
|
with self.assertRaisesRegex(ValidationError, "impersonate generated VM IDs"):
|
|
parse_name("vm-" + suffix)
|
|
for suffix in ("f" * 31, "f" * 33):
|
|
self.assertEqual(parse_name("vm-" + suffix), "vm-" + suffix)
|
|
|
|
def test_ports_accept_repeated_comma_lists_whitespace_and_case_insensitive_labels(self) -> None:
|
|
values = ("443:https,22", " 80:HtTp ,\t8080:tcp ", "65535, 1")
|
|
expected = (
|
|
PortRule(1), PortRule(22), PortRule(80, "HTTP"), PortRule(443, "HTTPS"),
|
|
PortRule(8080), PortRule(65535),
|
|
)
|
|
self.assertEqual(parse_ports(values), expected)
|
|
self.assertEqual(parse_ports(list(values)), expected)
|
|
|
|
def test_bare_ports_are_tcp_and_empty_sequences_do_not_add_defaults(self) -> None:
|
|
self.assertEqual(parse_ports(("80,443",)), (PortRule(80, "TCP"), PortRule(443, "TCP")))
|
|
self.assertEqual(parse_ports([]), ())
|
|
self.assertEqual(parse_ports(()), ())
|
|
|
|
def test_identical_port_labels_are_deduplicated_across_options_and_comma_lists(self) -> None:
|
|
self.assertEqual(
|
|
parse_ports(("22", "22:tcp", "80:HTTP,80:hTtP", "443:HTTPS", "443:https")),
|
|
(PortRule(22), PortRule(80, "HTTP"), PortRule(443, "HTTPS")),
|
|
)
|
|
|
|
def test_conflicting_tcp_service_labels_are_rejected(self) -> None:
|
|
for values in (
|
|
("80", "80:HTTP"), ("80:HTTP,80:TCP",),
|
|
("80:HTTP", "80:HTTPS"), ("80:https,80:http",),
|
|
):
|
|
with self.subTest(values=values):
|
|
with self.assertRaisesRegex(ValidationError, "conflicting labels for TCP port 80"):
|
|
parse_ports(values)
|
|
|
|
def test_ports_require_strict_port_colon_label_syntax(self) -> None:
|
|
for value in (
|
|
"", ",", "22,", ",22", "22,,80", "22, ,80", "TCP", "HTTP", "HTTPS",
|
|
"TCP:80", "80/tcp", ":HTTP", "80:", "80::HTTP", "80 :HTTP", "80: HTTP",
|
|
"80:HTTP S", "80:UDP", "80-81", "80.0", "+80", "0x50", "22 80", "80;id",
|
|
):
|
|
with self.subTest(value=value), self.assertRaises(ValidationError):
|
|
parse_ports((value,))
|
|
|
|
def test_parsed_ports_reject_out_of_range_numbers_and_non_integer_text(self) -> None:
|
|
for value in ("-1", "0", "65536", "999999999999", "True", "False", "None", "1.5"):
|
|
with self.subTest(value=value), self.assertRaises(ValidationError):
|
|
parse_ports((value,))
|
|
|
|
def test_ports_reject_scalar_strings_instead_of_splitting_them_into_digits(self) -> None:
|
|
for values in ("22", "", b"22"):
|
|
with self.subTest(values=values), self.assertRaises(ValidationError):
|
|
parse_ports(values)
|
|
|
|
def test_ports_require_a_sequence_not_a_scalar_mapping_or_set(self) -> None:
|
|
for values in (None, True, 22, {"22": "HTTP"}, {"22"}):
|
|
with self.subTest(values=values), self.assertRaises(ValidationError):
|
|
parse_ports(values)
|
|
|
|
def test_ports_require_string_sequence_members(self) -> None:
|
|
for value in (None, True, False, 22, 22.0, b"22", [], {}, PortRule(22)):
|
|
with self.subTest(value=value), self.assertRaises(ValidationError):
|
|
parse_ports(("80:HTTP", value))
|
|
|
|
def test_port_rules_require_real_integers_within_tcp_port_bounds(self) -> None:
|
|
self.assertEqual(PortRule(1).protocol, "TCP")
|
|
self.assertEqual(PortRule(65535, "https").protocol, "HTTPS")
|
|
for port in (None, True, False, 0, -1, 65536, 22.0, "22", b"22", [], {}):
|
|
with self.subTest(port=port), self.assertRaises(ValidationError):
|
|
PortRule(port)
|
|
|
|
def test_port_rule_labels_are_normalized_but_not_coerced_or_trimmed(self) -> None:
|
|
self.assertEqual(PortRule(8080, "hTtP"), PortRule(8080, "HTTP"))
|
|
for protocol in (None, True, 1, "", "UDP", "SCTP", " TCP", "TCP ", "tcp\n", [], {}):
|
|
with self.subTest(protocol=protocol), self.assertRaises(ValidationError):
|
|
PortRule(22, protocol)
|
|
|
|
def test_timeout_accepts_zero_and_fractional_seconds(self) -> None:
|
|
self.assertEqual(parse_timeout("0"), 0)
|
|
self.assertEqual(parse_timeout("0.5"), 0.5)
|
|
|
|
def test_timeout_rejects_negative_non_finite_or_invalid_values(self) -> None:
|
|
for value in ("-1", "nan", "inf", "-inf", "", "seconds", None):
|
|
with self.subTest(value=value), self.assertRaises(ValidationError):
|
|
parse_timeout(value)
|