From 17016e90cd32c20e0c86ee5de6c47841c2ab3a4a Mon Sep 17 00:00:00 2001 From: catborise Date: Tue, 6 Oct 2026 22:10:19 +0300 Subject: [PATCH] fix(computes): a compute's hostname and login cannot inject a connection URI or ssh option A compute's hostname and login go unquoted into the libvirt connection URI (qemu+ssh://login@host/system, qemu+tcp://host/system, qemu+tls://...) and, for an SSH compute, into the ssh command line of the DRBD status view. They were barely validated: validate_hostname used re.match with a negated class, so it caught only a leading special character, and login was never checked. A hostname like "192.0.2.1/system?command=/path" injected a libvirt URI parameter (verified: connecting ran the named command through the ssh wrapper), and TLS/SSH parameters such as no_verify, pkipath, socket or netcat could be added the same way; a login like "-oProxyCommand=..." became an ssh option. The DRF ComputeSerializer applied neither check, and validate_name tested a string literal instead of its argument, so it never raised. validate_hostname now accepts only an IPv4 address or a host name with fullmatch (no scheme, port, path, query, whitespace or IPv6 literal; IPv6 is left out until the URI building brackets it). validate_login accepts an empty value (TCP falls back to the server's authentication, the socket transport ignores the login) or a plain name with no leading "-". Both sit on the Compute model fields, so the forms and the DRF serializer (which copies model validators onto its fields) reject the values; a migration records the metadata. validate_name checks its argument. drbd_status validates the login and hostname before building login@host and ends ssh option parsing with "--"; an existing or bypassed bad row is skipped rather than run. Tests: hostname/login validators accept real values and reject URI and option injection; the SSH form and the serializer (create, and an empty TCP login) enforce them; drbd_status skips an injecting target. Checked the real test computes still validate. The new tests fail on the previous code. --- ...er_compute_hostname_alter_compute_login.py | 33 ++++++++++++ computes/models.py | 6 +-- computes/test_validators.py | 53 +++++++++++++++++++ computes/tests.py | 40 ++++++++++++++ computes/validators.py | 42 +++++++++------ instances/test_drbd_target.py | 17 ++++++ instances/views.py | 19 +++++-- 7 files changed, 189 insertions(+), 21 deletions(-) create mode 100644 computes/migrations/0005_alter_compute_hostname_alter_compute_login.py create mode 100644 computes/test_validators.py create mode 100644 instances/test_drbd_target.py diff --git a/computes/migrations/0005_alter_compute_hostname_alter_compute_login.py b/computes/migrations/0005_alter_compute_hostname_alter_compute_login.py new file mode 100644 index 00000000..9a2a7592 --- /dev/null +++ b/computes/migrations/0005_alter_compute_hostname_alter_compute_login.py @@ -0,0 +1,33 @@ +# Generated by Django 5.2.17 on 2026-10-06 19:07 + +import computes.validators +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("computes", "0004_compute_migration_address"), + ] + + operations = [ + migrations.AlterField( + model_name="compute", + name="hostname", + field=models.CharField( + max_length=64, + validators=[computes.validators.validate_hostname], + verbose_name="hostname", + ), + ), + migrations.AlterField( + model_name="compute", + name="login", + field=models.CharField( + blank=True, + max_length=20, + validators=[computes.validators.validate_login], + verbose_name="login", + ), + ), + ] diff --git a/computes/models.py b/computes/models.py index 67896a98..2d624e29 100644 --- a/computes/models.py +++ b/computes/models.py @@ -5,15 +5,15 @@ from django.utils.functional import cached_property from django.utils.translation import gettext_lazy as _ from libvirt import virConnect -from computes.validators import validate_migration_address +from computes.validators import validate_hostname, validate_login, validate_migration_address from vrtManager.connection import connection_manager from vrtManager.hostdetails import wvmHostDetails class Compute(Model): name = CharField(_("name"), max_length=64, unique=True) - hostname = CharField(_("hostname"), max_length=64) - login = CharField(_("login"), max_length=20) + hostname = CharField(_("hostname"), max_length=64, validators=[validate_hostname]) + login = CharField(_("login"), max_length=20, blank=True, validators=[validate_login]) password = CharField(_("password"), max_length=14, blank=True, null=True) details = CharField(_("details"), max_length=64, null=True, blank=True) type = IntegerField() diff --git a/computes/test_validators.py b/computes/test_validators.py new file mode 100644 index 00000000..178fcc8e --- /dev/null +++ b/computes/test_validators.py @@ -0,0 +1,53 @@ +"""A compute's hostname and login go unquoted into libvirt connection URIs +and an ssh command line; they are validated so neither can inject a URI +parameter (e.g. ?command=, ?no_verify=) or an ssh option.""" + +import unittest + +from django.core.exceptions import ValidationError + +from computes.validators import validate_hostname, validate_login, validate_name + + +class HostnameValidatorTestCase(unittest.TestCase): + def test_accepts_ipv4_and_host_names(self): + for value in ("192.0.2.10", "localhost", "host.example.com", "a-b.example.", "host1"): + with self.subTest(value=value): + validate_hostname(value) + + def test_rejects_uri_and_option_injection(self): + for value in ( + "-x", + "a b;c", + "host:2222", + "192.0.2.1/system?command=/evil.sh&a=", + "host/system?no_verify=1", + "qemu+ssh://host/system", + "2001:db8::1", + "", + "host name", + ): + with self.subTest(value=value): + with self.assertRaises(ValidationError): + validate_hostname(value) + + +class LoginValidatorTestCase(unittest.TestCase): + def test_accepts_plain_logins_and_empty(self): + # empty: TCP falls back to server auth, the socket transport ignores it + for value in ("", "root", "libvirt.user", "user-1", "_svc"): + with self.subTest(value=value): + validate_login(value) + + def test_rejects_option_and_delimiter_injection(self): + for value in ("-oProxyCommand=x", "u@realm", "a/b", "a b", "a\\b", "a%b"): + with self.subTest(value=value): + with self.assertRaises(ValidationError): + validate_login(value) + + +class NameValidatorTestCase(unittest.TestCase): + def test_checks_the_value(self): + validate_name("good-name_1.2") + with self.assertRaises(ValidationError): + validate_name("bad name!") diff --git a/computes/tests.py b/computes/tests.py index accfbf4b..df3be415 100644 --- a/computes/tests.py +++ b/computes/tests.py @@ -317,3 +317,43 @@ class ComputeConcurrencyTestCase(TransactionTestCase): with libvirt_compute_lock(compute, timeout=1.0): executed.append("deep") self.assertEqual(executed, ["outer", "inner", "deep"]) + + +class ComputeInputValidationTestCase(TestCase): + """hostname and login reach libvirt URIs and ssh unquoted; the form and + the API reject values that would inject a URI parameter or an ssh option.""" + + def test_form_rejects_injecting_hostname(self): + from computes.forms import SshComputeForm + + form = SshComputeForm( + data={"name": "c", "hostname": "192.0.2.1/system?command=/evil", "login": "root", "type": 2} + ) + self.assertFalse(form.is_valid()) + self.assertIn("hostname", form.errors) + + def test_api_rejects_injecting_hostname_and_login(self): + from computes.api.serializers import ComputeSerializer + + for field, data in ( + ("hostname", {"name": "c", "hostname": "192.0.2.1/system?command=/evil&a=", "login": "root", "type": 2}), + ("login", {"name": "c", "hostname": "192.0.2.1", "login": "-oProxyCommand=x", "type": 2}), + ): + with self.subTest(field=field): + serializer = ComputeSerializer(data=data) + self.assertFalse(serializer.is_valid()) + self.assertIn(field, serializer.errors) + + def test_api_accepts_valid_values(self): + from computes.api.serializers import ComputeSerializer + + serializer = ComputeSerializer( + data={"name": "c", "hostname": "host.example.com", "login": "root", "type": 2} + ) + self.assertTrue(serializer.is_valid(), serializer.errors) + + def test_api_accepts_empty_login_for_tcp(self): + from computes.api.serializers import ComputeSerializer + + serializer = ComputeSerializer(data={"name": "c", "hostname": "192.0.2.1", "login": "", "type": 1}) + self.assertTrue(serializer.is_valid(), serializer.errors) diff --git a/computes/validators.py b/computes/validators.py index ee5bc5b2..7c6080cc 100644 --- a/computes/validators.py +++ b/computes/validators.py @@ -4,30 +4,42 @@ from ipaddress import ip_address from django.core.exceptions import ValidationError from django.utils.translation import gettext_lazy as _ -have_symbol = re.compile("[^a-zA-Z0-9._-]+") -wrong_ip = re.compile("^0.|^255.") -wrong_name = re.compile("[^a-zA-Z0-9._-]+") +# one or more dot-separated RFC 1123 labels, optional trailing dot +_hostname = re.compile(r"[A-Za-z0-9]([A-Za-z0-9-]{0,61}[A-Za-z0-9])?(\.[A-Za-z0-9]([A-Za-z0-9-]{0,61}[A-Za-z0-9])?)*\.?") +_login = re.compile(r"[A-Za-z0-9._][A-Za-z0-9._-]*") +_name = re.compile(r"[A-Za-z0-9._-]+") def validate_hostname(value): - sym = have_symbol.match(value) - wip = wrong_ip.match(value) + """An IPv4 address or a host name that goes safely into a libvirt + connection URI. No scheme, port, path, query, whitespace or IPv6 literal: + those would change the URI (an injected ?command=, ?no_verify=, a port or + a userinfo separator). IPv6 is rejected until the URI building brackets + it.""" + try: + if ip_address(value).version == 4: + return + except ValueError: + if _hostname.fullmatch(value): + return + raise ValidationError(_("Enter an IPv4 address or a host name, without a scheme, port or path")) - if sym: + +def validate_login(value): + """A user name that goes safely into a libvirt connection URI and an ssh + command line. Empty is allowed (TCP falls back to the server's own + authentication, the socket transport ignores the login); otherwise only + letters, digits, ".", "_" and "-", and never a leading "-" (an ssh + option).""" + if value and not _login.fullmatch(value): raise ValidationError( - _('Hostname must contain only numbers, or the domain name separated by "."') + _("The login may contain only letters, digits, '.', '_' and '-', and may not start with '-'") ) - elif wip: - raise ValidationError(_("Wrong IP address")) def validate_name(value): - have_symbol = wrong_name.match("[^a-zA-Z0-9._-]+") - if have_symbol: - raise ValidationError(_("The hostname must not contain any special characters")) - - -_hostname = re.compile(r"[A-Za-z0-9]([A-Za-z0-9-]{0,61}[A-Za-z0-9])?(\.[A-Za-z0-9]([A-Za-z0-9-]{0,61}[A-Za-z0-9])?)*\.?") + if not _name.fullmatch(value): + raise ValidationError(_("The name must not contain any special characters")) def validate_migration_address(value): diff --git a/instances/test_drbd_target.py b/instances/test_drbd_target.py new file mode 100644 index 00000000..5c29cd15 --- /dev/null +++ b/instances/test_drbd_target.py @@ -0,0 +1,17 @@ +"""drbd_status runs ssh with the compute's login@hostname; a compute whose +values would inject an ssh option (an existing or bypassed row) is skipped.""" + +import unittest +from types import SimpleNamespace + +from instances.views import _valid_ssh_target + + +class DrbdTargetTestCase(unittest.TestCase): + def test_valid_target(self): + self.assertTrue(_valid_ssh_target(SimpleNamespace(login="root", hostname="192.0.2.1"))) + + def test_option_injection_is_skipped(self): + for login, hostname in (("-oProxyCommand=x", "192.0.2.1"), ("root", "-x"), ("root", "a b")): + with self.subTest(login=login, hostname=hostname): + self.assertFalse(_valid_ssh_target(SimpleNamespace(login=login, hostname=hostname))) diff --git a/instances/views.py b/instances/views.py index d2ae2040..e442a57e 100755 --- a/instances/views.py +++ b/instances/views.py @@ -16,12 +16,13 @@ from admin.decorators import superuser_only from appsettings.models import AppSettings from appsettings.settings import app_settings from computes.models import Compute +from computes.validators import validate_hostname, validate_login from computes.utils import libvirt_compute_lock, libvirt_instance_lock, user_quota_lock from django.conf import settings from django.contrib import messages from django.contrib.auth.decorators import permission_required from django.contrib.auth.models import User -from django.core.exceptions import PermissionDenied +from django.core.exceptions import PermissionDenied, ValidationError from django.http import Http404, HttpResponse, JsonResponse from django.db.models import prefetch_related_objects from django.shortcuts import get_object_or_404, redirect, render @@ -194,14 +195,26 @@ def status(request, pk): return JsonResponse({"status": instance.proxy.get_status()}) +def _valid_ssh_target(compute): + """The login and hostname reach ssh as a command-line argument; a compute + whose values would inject an ssh option (an existing or bypassed row that + the field validators did not cover) is skipped rather than run.""" + try: + validate_login(compute.login) + validate_hostname(compute.hostname) + except ValidationError: + return False + return True + + def drbd_status(request, pk): instance = get_instance(request.user, pk) result = "None DRBD" - if instance.compute.type == 2: + if instance.compute.type == 2 and _valid_ssh_target(instance.compute): conn = instance.compute.login + "@" + instance.compute.hostname remoteDrbdStatus = subprocess.run( - ["ssh", conn, "sudo", "/usr/sbin/drbdadm", "status", "&&", "exit"], + ["ssh", "--", conn, "sudo", "/usr/sbin/drbdadm", "status"], stdout=subprocess.PIPE, text=True, )