mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-08 19:26:48 +00:00
890d82f44a
When the guest agent does not answer a root password or SSH key change in time, the command was already sent: the agent runs it when it answers again. Measured: with the agent frozen the change failed with "guest agent command timed out", and after it resumed, root's hash was the one sent. The error now adds "The change may still be made when the agent answers again." A failed synchronization sends no command and gets no note.
221 lines
11 KiB
Python
221 lines
11 KiB
Python
import json
|
|
import socket
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from django.contrib.messages import get_messages
|
|
from django.test import RequestFactory, TestCase
|
|
from django.urls import reverse
|
|
|
|
from computes.models import Compute
|
|
from instances.models import Instance
|
|
from libvirt import libvirtError
|
|
|
|
|
|
class SetRootPassTestCase(TestCase):
|
|
def setUp(self):
|
|
self.client.login(request=RequestFactory().get("/"), username="admin", password="admin")
|
|
self.compute = Compute.objects.create(
|
|
name="test_compute_root_pass",
|
|
hostname="localhost",
|
|
type=1,
|
|
)
|
|
self.instance = Instance.objects.create(
|
|
compute=self.compute,
|
|
name="root-pass-vm",
|
|
uuid="87654321-4321-4321-4321-210987654321",
|
|
)
|
|
|
|
@patch("socket.socket")
|
|
@patch.object(Instance, "proxy")
|
|
def test_set_root_pass_post(self, mock_proxy, mock_socket_cls):
|
|
# Renamed on the host: gstfsd must get the name of this UUID's domain
|
|
mock_proxy.instance.name.return_value = "renamed-on-host"
|
|
mock_proxy.get_status.return_value = 5 # shut off
|
|
mock_sock = MagicMock()
|
|
mock_socket_cls.return_value = mock_sock
|
|
replies = iter([json.dumps({"return": "success"}).encode(), b""] * 2)
|
|
# one reply, then gstfsd closes the connection
|
|
mock_sock.recv.side_effect = lambda size: next(replies)
|
|
|
|
response = self.client.post(
|
|
reverse("instances:rootpasswd", args=[self.instance.id]),
|
|
{"passwd": "newSecretPassword123"}
|
|
)
|
|
# Redirects back to referer or instance detail
|
|
self.assertEqual(response.status_code, 302)
|
|
|
|
# Check socket sent the command with hashed password
|
|
self.assertTrue(mock_sock.sendall.called)
|
|
sent_bytes = mock_sock.sendall.call_args[0][0]
|
|
data = json.loads(sent_bytes.decode())
|
|
self.assertEqual(data["action"], "password")
|
|
self.assertEqual(data["vname"], "renamed-on-host")
|
|
# SHA-512 crypt with a random salt, not the old fixed one
|
|
self.assertTrue(data["passwd"].startswith("$6$"))
|
|
self.assertFalse(data["passwd"].startswith("$6$kgPoiREy$"))
|
|
|
|
def test_set_root_pass_get_not_allowed(self):
|
|
response = self.client.get(reverse("instances:rootpasswd", args=[self.instance.id]))
|
|
self.assertEqual(response.status_code, 405)
|
|
|
|
def test_set_root_pass_not_found(self):
|
|
response = self.client.post(reverse("instances:rootpasswd", args=[99999]), {"passwd": "secret"})
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
@patch("socket.socket")
|
|
@patch.object(Instance, "proxy")
|
|
def test_same_password_gets_a_different_salt_each_time(self, mock_proxy, mock_socket_cls):
|
|
mock_proxy.instance.name.return_value = "root-pass-vm"
|
|
mock_proxy.get_status.return_value = 5
|
|
mock_sock = MagicMock()
|
|
mock_socket_cls.return_value = mock_sock
|
|
replies = iter([json.dumps({"return": "success"}).encode(), b""] * 2)
|
|
# one reply, then gstfsd closes the connection
|
|
mock_sock.recv.side_effect = lambda size: next(replies)
|
|
url = reverse("instances:rootpasswd", args=[self.instance.id])
|
|
|
|
self.client.post(url, {"passwd": "samePassword1"})
|
|
self.client.post(url, {"passwd": "samePassword1"})
|
|
|
|
hashes = [json.loads(c[0][0].decode())["passwd"] for c in mock_sock.sendall.call_args_list]
|
|
self.assertEqual(len(hashes), 2)
|
|
self.assertNotEqual(hashes[0], hashes[1])
|
|
|
|
@patch("socket.socket")
|
|
@patch.object(Instance, "proxy")
|
|
def test_unreachable_gstfsd_is_an_error_message_not_a_crash(self, mock_proxy, mock_socket_cls):
|
|
mock_proxy.instance.name.return_value = "root-pass-vm"
|
|
mock_proxy.get_status.return_value = 5
|
|
mock_sock = MagicMock()
|
|
mock_socket_cls.return_value = mock_sock
|
|
mock_sock.connect.side_effect = socket.timeout("timed out")
|
|
|
|
response = self.client.post(
|
|
reverse("instances:rootpasswd", args=[self.instance.id]), {"passwd": "x"}
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 302)
|
|
mock_sock.settimeout.assert_called()
|
|
|
|
@patch("socket.socket")
|
|
@patch.object(Instance, "proxy")
|
|
def test_add_public_key_unreachable_gstfsd_is_an_error_message(self, mock_proxy, mock_socket_cls):
|
|
from accounts.models import UserSSHKey
|
|
from django.contrib.auth import get_user_model
|
|
|
|
admin = get_user_model().objects.get(username="admin")
|
|
key = UserSSHKey.objects.create(user=admin, keyname="k", keypublic="ssh-ed25519 AAAA k")
|
|
mock_proxy.instance.name.return_value = "root-pass-vm"
|
|
mock_proxy.get_status.return_value = 5
|
|
mock_sock = MagicMock()
|
|
mock_socket_cls.return_value = mock_sock
|
|
mock_sock.connect.side_effect = ConnectionRefusedError()
|
|
|
|
response = self.client.post(
|
|
reverse("instances:add_public_key", args=[self.instance.id]), {"sshkeyid": key.id}
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 302)
|
|
mock_sock.settimeout.assert_called()
|
|
|
|
|
|
class GuestAgentAccessTestCase(TestCase):
|
|
"""A running VM whose guest agent answers gets its root password and SSH
|
|
key through the agent; gstfsd serves only a shut-off VM."""
|
|
|
|
def setUp(self):
|
|
from accounts.models import UserSSHKey
|
|
from django.contrib.auth import get_user_model
|
|
|
|
self.client.login(request=RequestFactory().get("/"), username="admin", password="admin")
|
|
compute = Compute.objects.create(name="agent-compute", hostname="192.0.2.1", type=1)
|
|
self.instance = Instance.objects.create(compute=compute, name="agent-vm",
|
|
uuid="11111111-2222-3333-4444-555555555555")
|
|
admin = get_user_model().objects.get(username="admin")
|
|
self.key = UserSSHKey.objects.create(user=admin, keyname="laptop", keypublic="ssh-ed25519 AAAA laptop")
|
|
patcher = patch.object(Instance, "proxy")
|
|
self.proxy = patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
self.proxy.get_status.return_value = 1 # running
|
|
self.proxy.is_agent_ready.return_value = True
|
|
|
|
def post(self, name, data):
|
|
with patch("instances.views.gstfsd_request") as gstfsd:
|
|
response = self.client.post(reverse(f"instances:{name}", args=[self.instance.id]), data)
|
|
gstfsd.assert_not_called()
|
|
return [str(m) for m in get_messages(response.wsgi_request)]
|
|
|
|
def test_the_agent_sets_the_root_password(self):
|
|
self.assertEqual(self.post("rootpasswd", {"passwd": "S3cret-pass"}), ["Reset root password"])
|
|
user, crypted = self.proxy.set_user_password.call_args.args
|
|
self.assertEqual(user, "root")
|
|
self.assertTrue(crypted.startswith("$6$"))
|
|
|
|
def test_the_agent_adds_the_ssh_key(self):
|
|
self.assertEqual(self.post("add_public_key", {"sshkeyid": self.key.id}),
|
|
["Installed new SSH public key laptop"])
|
|
self.proxy.add_authorized_key.assert_called_once_with("root", "ssh-ed25519 AAAA laptop")
|
|
|
|
def test_a_running_vm_without_its_agent_is_refused(self):
|
|
self.proxy.is_agent_ready.return_value = False
|
|
for name, data in (("rootpasswd", {"passwd": "S3cret-pass"}), ("add_public_key", {"sshkeyid": self.key.id})):
|
|
# the latest message: the redirect is not followed, so earlier ones stay
|
|
self.assertEqual(self.post(name, data)[-1], "Shut the VM down, or start its QEMU guest agent, and try again")
|
|
self.proxy.set_user_password.assert_not_called()
|
|
self.proxy.add_authorized_key.assert_not_called()
|
|
|
|
def test_an_agent_that_refuses_is_an_error_message(self):
|
|
self.proxy.set_user_password.side_effect = libvirtError("guest agent command failed: user root not found")
|
|
self.assertEqual(self.post("rootpasswd", {"passwd": "S3cret-pass"}),
|
|
["guest agent command failed: user root not found"])
|
|
|
|
def test_a_command_that_timed_out_may_still_be_made(self):
|
|
# sent to the agent: it runs when the agent answers again; a failed
|
|
# synchronization sends nothing
|
|
from libvirt import VIR_ERR_AGENT_UNRESPONSIVE
|
|
|
|
for code, text, note in (
|
|
(112, "guest agent command timed out", True), # VIR_ERR_AGENT_COMMAND_TIMEOUT
|
|
(VIR_ERR_AGENT_UNRESPONSIVE, "Guest agent is not responding", False),
|
|
):
|
|
with self.subTest(code=code):
|
|
err = libvirtError(text)
|
|
err.get_error_code = lambda code=code: code
|
|
self.proxy.set_user_password.side_effect = err
|
|
self.proxy.add_authorized_key.side_effect = err
|
|
expected = text + (" The change may still be made when the agent answers again." if note else "")
|
|
self.assertEqual(self.post("rootpasswd", {"passwd": "S3cret-pass"})[-1], expected)
|
|
self.assertEqual(self.post("add_public_key", {"sshkeyid": self.key.id})[-1], expected)
|
|
|
|
def test_a_user_who_does_not_see_hosts_gets_no_connection_details(self):
|
|
from accounts.models import UserInstance
|
|
from django.contrib.auth import get_user_model
|
|
from libvirt import VIR_ERR_SYSTEM_ERROR, VIR_FROM_RPC
|
|
|
|
owner = get_user_model().objects.create_user("agent-owner", password="pw")
|
|
UserInstance.objects.create(user=owner, instance=self.instance, is_change=True)
|
|
self.client.force_login(owner)
|
|
err = libvirtError("Cannot recv data: 192.0.2.1:16509 Connection reset by peer")
|
|
err.get_error_domain = lambda: VIR_FROM_RPC
|
|
err.get_error_code = lambda: VIR_ERR_SYSTEM_ERROR
|
|
self.proxy.set_user_password.side_effect = err
|
|
self.assertEqual(self.post("rootpasswd", {"passwd": "S3cret-pass"}),
|
|
["The host of this virtual machine cannot be reached right now."])
|
|
|
|
|
|
class OsInfoErrorTestCase(TestCase):
|
|
"""The VM page asks for the guest's OS details; an agent that does not
|
|
answer in time is an error answer, not a server error."""
|
|
|
|
def setUp(self):
|
|
self.client.login(request=RequestFactory().get("/"), username="admin", password="admin")
|
|
compute = Compute.objects.create(name="osinfo-compute", hostname="192.0.2.1", type=1)
|
|
self.instance = Instance.objects.create(compute=compute, name="osinfo-vm",
|
|
uuid="22222222-3333-4444-5555-666666666666")
|
|
|
|
@patch.object(Instance, "proxy")
|
|
def test_a_silent_agent_is_an_error_answer(self, proxy):
|
|
proxy.osinfo.side_effect = libvirtError("guest agent didn't respond to command within '10' seconds")
|
|
response = self.client.get(reverse("instances:osinfo", args=[self.instance.id]))
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertEqual(response.json(), {"error": "guest agent didn't respond to command within '10' seconds"})
|